Building A Distributed Actor Framework On Top Of Tcp: Messaging, Supervision, And Location Transparency

This is the reality that motivated the actor model. Born from Carl Hewitt’s work in the 1970s and later refined by the Erlang community, the actor model offers a different mental model for concurrency and distribution: instead of shared state and locks, you have independent, encapsulated units — actors — that communicate exclusively through asynchronous messages. Each actor has a mailbox, a behavior, and the ability to spawn child actors. Crucially, actors are location‑transparent — an actor’s address carries no information about whether it resides in the same process, on the same machine, or across a continent. This abstraction decouples the *what* (the actor’s logic) from the *where* (its physical or network location). The result is a framework that tames the chaos of distributed computing by making failure part of the design, not an afterthought.

Yet, for all the elegance of the actor model, implementing a *distributed* actor system that works reliably over a network is a profoundly challenging engineering task. The actor model abstracts away the network, but the network does not abstract away its own realities. Packets are dropped, delayed, duplicated, or reordered. Connections fail, machines crash, and clocks drift. You cannot simply serialize an actor’s mailbox and ship it over TCP—the semantics of message delivery, ordering, and exactly‑once processing become murky the moment you leave a single address space. The clean, single‑node actor system collapses under the weight of network partitions, split‑brain scenarios, and the herculean effort of maintaining a consistent global state across unreliable links.

In this series, we will roll up our sleeves and build a distributed actor framework from scratch, using TCP as the bedrock transport layer. TCP offers a reliable, ordered, stream‑oriented connection — which simplifies some problems (no need to worry about packet reordering within a single connection) but introduces others (head‑of‑line blocking, stream boundaries, connection lifecycle). We will explore how to wrap TCP with an actor‑friendly protocol, how to handle remote actor references, how to implement supervision across machine boundaries, and how to test the whole thing under realistic failure conditions. By the end, you will have a deep understanding of the tradeoffs involved in building distributed systems, and a working prototype that you can extend for your own projects.

---

## 1. Core Actor Model Concepts – A Refresher

Before we dive into the network layer, we must have a solid grasp of the actor model as it exists on a single node. Many of the design decisions we make for distribution are driven by the desire to preserve the semantics that make actors appealing in the first place.

**Actors as fine‑grained units of computation.** An actor is an entity that encapsulates state and behavior. It has a mailbox — a queue of messages — and it processes those messages one at a time, in the order they are received (FIFO per actor). While processing a message, the actor can:
- Send messages to other actors (whose addresses it knows).
- Create new actors (child actors).
- Change its own behavior for subsequent messages.

Because actors are single‑threaded by design, you never need locks or mutexes inside an actor. This eliminates a huge class of concurrency bugs.

**Actor references and addresses.** In a single‑node system, an actor reference is often just a pointer or an integer ID. But for distribution, we need a globally unique address that includes information about the node where the actor lives. A common scheme is `actor://host:port/path/to/actor`. The address encodes the transport layer (TCP), the network endpoint, and a hierarchical path.

**Mailbox semantics.** On a single node, the mailbox is an in‑memory queue. When a message is sent to an actor, it is appended to the tail of the queue. The actor’s scheduler picks up the next message and runs the actor’s behavior on it. This is fast and deterministic. In a distributed setting, the mailbox is split: the sending side places the message into a network buffer; the receiving side deserializes and enqueues it into the local mailbox. The network introduces latency, possible reordering (if multiple connections are used), and the risk of message loss.

**Supervision and fault tolerance.** In Erlang/OTP and similar frameworks, actors are organized into a supervision tree. A supervisor actor monitors its child actors and restarts them if they fail. This creates a hierarchical error‑recovery mechanism: a failing leaf actor is restarted by its direct supervisor; if the supervisor itself crashes, its own supervisor takes action. This design isolates failures and prevents cascading crashes. Distributing supervision across machines is non‑trivial: how does a supervisor on node A know that an actor on node B has crashed? How does it restart it if node B is unreachable?

**Location transparency.** The promise of the actor model is that you can send a message to any actor without caring where it lives. The runtime handles routing. This means the same code that sends a message to a local actor should work unchanged when the target is on a remote node. Achieving this requires a unified addressing scheme and a transparent layer that decides whether to deliver the message locally or serialize it and push it over TCP.

With these concepts fresh in mind, we can now examine how TCP fits into the picture.

---

## 2. Why TCP? The Case for a Reliable Stream

When you think about building a distributed system, you have a spectrum of transport options: raw UDP, TCP, QUIC, SCTP, or even higher‑level messaging protocols like AMQP or MQTT. For our actor framework, we choose TCP as the foundation. Why?

**Reliability without reinventing the wheel.** TCP guarantees in‑order delivery of bytes, retransmits lost packets, and handles congestion control. Without TCP, we would have to implement our own acknowledgement and retry logic, sequence numbers, and flow control. That is an enormous amount of work that would distract from the core actor protocol. By standing on TCP’s shoulders, we can focus on the actor‑specific concerns: message framing, routing, and failure detection.

**Stream abstraction vs. message boundaries.** One of the first hurdles when using TCP for message‑oriented communication is that TCP is a stream protocol. It does not preserve message boundaries. If you send a 100‑byte message followed by a 200‑byte message, the receiver may read 150 bytes in one `recv()` call and 150 in the next. To use TCP for actors, we must implement a **frame protocol** — a way to delimit messages within the byte stream. Common approaches include:
- **Length‑prefixed framing:** write a fixed‑size integer (e.g., 4 bytes) that indicates the length of the following message, then the message bytes.
- **Delimiter‑based framing:** use a special byte sequence to mark the end of a message (e.g., newline for text‑based protocols).
- **Fixed‑size messages:** not practical for variable‑length data.

We will use length‑prefixed framing because it is efficient and works well with binary serialization.

**Connection lifecycle.** TCP requires connection establishment (three‑way handshake) and teardown. In a distributed actor system, we typically maintain persistent connections between nodes to avoid the overhead of repeated handshakes. However, connections can break due to network failures or timeouts. We need to detect broken connections, attempt reconnection, and handle the transient state where a node is unreachable but may come back.

**Head‑of‑line blocking.** Since TCP delivers bytes in order, all messages sent on the same connection are serialized. If a large message takes a long time to serialize or deserialize, it blocks subsequent smaller messages. This is a well‑known downside. We can mitigate it by using multiple connections between nodes, or by implementing a multiplexing layer on top of a single connection (like HTTP/2 does). For simplicity, we will start with a single connection and later discuss multiplexing.

**Nagle’s algorithm and buffering.** By default, TCP uses Nagle’s algorithm to coalesce small packets into larger ones, which can introduce latency for small messages. For an actor system where many small messages are common, we may want to disable Nagle (`TCP_NODELAY`) to reduce latency. We will make this configuration explicit.

Given these properties, TCP provides a solid but imperfect foundation. The imperfections (head‑of‑line blocking, stream boundaries) must be addressed by our application layer. Let’s now design a protocol that sits on top of TCP.

---

## 3. Designing the Actor Protocol on TCP

Our protocol must enable:
- Reliable, ordered delivery of actor messages between nodes.
- Identification of source and destination actors.
- Support for different message types (user data, system messages, heartbeats).
- Serialization and deserialization of messages.

We will define a simple binary packet format:

- **Length:** total length of the packet (excluding the length field itself), so the receiver knows how many bytes to read for the rest.
- **Protocol version:** allows future protocol evolution.
- **Msg type:** distinguishes user data from control messages (heartbeat, acknowledgment, error, etc.).
- **Reserved:** for flags (e.g., priority, compression).
- **Dest actor ref** and **Src actor ref**: strings or binary IDs that uniquely identify the actor. We can use a URL‑like format like `actor://node_id/actor_path`.
- **Payload:** the serialized content of the actor message (e.g., JSON, Protocol Buffers, or a custom binary format).

**Serialization choices.** For demonstration, we will use a simple JSON‑based serialization for messages – it is easy to debug and implement. In production, you would likely use a more efficient binary format (Protobuf, MessagePack, Avro). The important thing is that the serialization is transparent to the actor code: the framework should handle it automatically.

**System messages.** Besides user messages, we need internal messages for the framework’s operation:
- `HEARTBEAT` – sent periodically to check if the remote node is alive.
- `ACK` – optional acknowledgment of message receipt (if we want reliable delivery).
- `ERROR` – error codes (e.g., unknown actor, deserialization failure).
- `MONITOR` – request to be notified if a remote actor terminates.
- `TERMINATE` – notification that an actor has stopped.

These system messages will have a specific msg type code, and the receiver’s framework will handle them transparently, not forwarding them to user‑defined behavior.

**Framing and parsing.** On the receiving side, we must read exactly the number of bytes specified by the length field. This requires reading from the TCP socket in a loop until we have the full packet. We will implement a state machine: read length (4 bytes), then read the remaining bytes. We buffer partial reads.

---

## 4. Implementing a Basic (Single‑Node) Actor System

Before we add networking, let’s implement a minimal actor system in a language of your choice. I’ll use Python for clarity, but the concepts translate to any language.

```python
import threading
import queue
import uuid

class ActorException(Exception):
    pass

class ActorRef:
    """Lightweight reference to an actor (local only for now)."""
    def __init__(self, actor_id, system):
        self.actor_id = actor_id
        self._system = system  # weak reference to the actor system

    def tell(self, message):
        self._system.send(self, message)

class Mailbox:
    def __init__(self):
        self.queue = queue.Queue()
        self._lock = threading.Lock()
        self._actor = None

    def attach(self, actor):
        self._actor = actor

    def enqueue(self, message):
        self.queue.put(message)

    def run(self):
        while True:
            msg = self.queue.get()
            if msg is None:  # poison pill
                break
            try:
                self._actor.on_message(msg)
            except Exception as e:
                # In a real system, we would invoke supervisor handling
                print(f"Actor {self._actor.actor_id} crashed: {e}")
                # For now, just stop
                break
        # cleanup

class Actor:
    def __init__(self, actor_id, system):
        self.actor_id = actor_id
        self.system = system
        self.mailbox = Mailbox()
        self.mailbox.attach(self)

    def on_message(self, message):
        raise NotImplementedError

class ActorSystem:
    def __init__(self):
        self._actors = {}
        self._lock = threading.Lock()

    def spawn(self, actor_class, *args, **kwargs):
        actor_id = str(uuid.uuid4())
        actor = actor_class(actor_id, self, *args, **kwargs)
        with self._lock:
            self._actors[actor_id] = actor
        t = threading.Thread(target=actor.mailbox.run)
        t.start()
        return ActorRef(actor_id, self)

    def send(self, ref, message):
        with self._lock:
            actor = self._actors.get(ref.actor_id)
        if actor:
            actor.mailbox.enqueue(message)
        else:
            raise ActorException(f"Actor {ref.actor_id} not found")

    def stop(self, ref):
        with self._lock:
            actor = self._actors.pop(ref.actor_id, None)
        if actor:
            actor.mailbox.enqueue(None)  # poison pill

class RemoteActorRef(ActorRef):
    def __init__(self, actor_id, node_manager, remote_node_host, remote_node_port):
        super().__init__(actor_id, node_manager)  # system is node_manager
        self.remote_node = (remote_node_host, remote_node_port)

    def tell(self, message):
        # Instead of local delivery, we ask node_manager to send over TCP
        self._system.send_remote(self, message)

class NodeManager:
    def __init__(self, local_host, local_port):
        self.local_host = local_host
        self.local_port = local_port
        self._connections = {}  # (host,port) -> TcpConnection
        self._local_actors = {}  # actor_id -> Actor
        self._listener = None

    def send_remote(self, ref, message):
        # Serialize
        data = serialize_message(ref.actor_id, message)
        target = ref.remote_node
        conn = self.get_connection(target)
        conn.send(data)

    def get_connection(self, addr):
        if addr not in self._connections:
            sock = socket.create_connection(addr)
            # Disable Nagle
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            conn = TcpConnection(sock)
            self._connections[addr] = conn
            # Start reading thread for this connection
            threading.Thread(target=self._reader, args=(conn,), daemon=True).start()
        return self._connections[addr]

def _handle_incoming(self, packet):
    dest_id, src_ref, msg_type, payload = deserialize_packet(packet)
    if dest_id in self._local_actors:
        actor = self._local_actors[dest_id]
        # Create a sender ref (possibly remote)
        sender = RemoteActorRef(src_ref.actor_id, self, src_ref.host, src_ref.port)
        # Wrap in a system message envelope?
        actor.mailbox.enqueue((sender, payload))
    else:
        # Could forward? For now, log error
        print(f"Actor {dest_id} not found locally")

# In NodeManager, when an actor terminates:
def _actor_terminated(self, actor_id, reason):
    # Notify all monitors (local and remote)
    monitors = self._monitors.get(actor_id, [])
    for monitor_ref in monitors:
        if isinstance(monitor_ref, RemoteActorRef):
            # Send TERMINATED message over TCP
            self.send_remote(monitor_ref, ('TERMINATED', actor_id, reason))
        else:
            # Local delivery
            local_actor = self._local_actors[monitor_ref.actor_id]
            local_actor.mailbox.enqueue(('TERMINATED', actor_id, reason))

def remote_spawn(self, remote_node, class_name, args, parent_ref):
    # Send SPAWN system message
    spawn_msg = ('SPAWN', class_name, args, parent_ref)
    self.send_system(remote_node, spawn_msg)
    # We'll handle the response asynchronously; store a future for the new ref

class ChatRoom(Actor):
    def __init__(self, actor_id, system):
        super().__init__(actor_id, system)
        self.subscribers = set()

    def on_message(self, sender, msg):
        if msg['type'] == 'Join':
            self.subscribers.add(sender)
            print(f"User joined: {msg['username']}")
        elif msg['type'] == 'Leave':
            self.subscribers.discard(sender)
        elif msg['type'] == 'ChatMessage':
            broadcast = {'from': msg['username'], 'text': msg['text']}
            for sub in self.subscribers:
                sub.tell(broadcast)

def test_remote_message():
    node1 = start_node('127.0.0.1', 9001)
    node2 = start_node('127.0.0.1', 9002)
    # Register node2 as peer of node1
    node1.connect_peer('127.0.0.1', 9002)
    node2.connect_peer('127.0.0.1', 9001)
    # Spawn echo actor on node1
    ref = node1.spawn(EchoActor)
    # Send from node2
    remote_ref = RemoteActorRef(ref.actor_id, node2, '127.0.0.1', 9001)
    remote_ref.tell("hello")
    time.sleep(0.5)
    assert node1.received_messages == ["hello"]  # some global state