# Reactor Module
The `erlang.reactor` module provides low-level FD-based protocol handling for building custom servers. It enables Python to implement protocol logic while Erlang handles I/O scheduling via `enif_select`.
## Overview
The reactor pattern separates I/O multiplexing (handled by Erlang) from protocol logic (handled by Python). This provides:
- **Efficient I/O** - Erlang's `enif_select` for event notification
- **Protocol flexibility** - Python implements the protocol state machine
- **Zero-copy buffers** - ReactorBuffer provides zero-copy data access via buffer protocol
- **Works with any fd** - TCP, UDP, Unix sockets, pipes, etc.
### Architecture
```
┌──────────────────────────────────────────────────────────────────────┐
│ Reactor Architecture │
├──────────────────────────────────────────────────────────────────────┤
│ │
│ Erlang (BEAM) Python │
│ ───────────── ────── │
│ │
│ ┌─────────────────────┐ ┌─────────────────────────────┐ │
│ │ py_reactor_context │ │ erlang.reactor │ │
│ │ │ │ │ │
│ │ accept() ──────────┼──fd_handoff─▶│ init_connection(fd, info) │ │
│ │ │ │ │ │ │
│ │ enif_select(READ) │ │ ▼ │ │
│ │ │ │ │ Protocol.connection_made() │ │
│ │ ▼ │ │ │ │
│ │ {select, fd, READ} │ │ │ │
│ │ │ │ │ │ │
│ │ └─────────────┼─on_read_ready│ Protocol.data_received() │ │
│ │ │ │ │ │ │
│ │ action = "write_ │◀─────────────┼───────┘ │ │
│ │ pending" │ │ │ │
│ │ │ │ │ │ │
│ │ enif_select(WRITE) │ │ │ │
│ │ │ │ │ │ │
│ │ ▼ │ │ │ │
│ │ {select, fd, WRITE}│ │ │ │
│ │ │ │ │ │ │
│ │ └─────────────┼on_write_ready│ Protocol.write_ready() │ │
│ │ │ │ │ │
│ └─────────────────────┘ └─────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────────────┘
```
## Protocol Base Class
The `Protocol` class is the base for implementing custom protocols:
```python
import erlang.reactor as reactor
class Protocol(reactor.Protocol):
"""Base protocol attributes and methods."""
# Set by reactor on connection
fd: int # File descriptor
client_info: dict # Connection metadata from Erlang
write_buffer: bytearray # Buffer for outgoing data
closed: bool # Whether connection is closed
```
### Lifecycle Methods
#### `connection_made(fd, client_info)`
Called when a file descriptor is handed off from Erlang.
```python
def connection_made(self, fd: int, client_info: dict):
"""Called when fd is handed off from Erlang.
Args:
fd: File descriptor for the connection
client_info: Dict with connection metadata
- 'addr': Client IP address
- 'port': Client port
- 'type': Connection type (tcp, udp, unix, etc.)
"""
# Initialize per-connection state
self.request_buffer = bytearray()
```
#### `data_received(data) -> action`
Called when data has been read from the fd. The `data` argument is a `ReactorBuffer` - a bytes-like object that supports zero-copy access via the buffer protocol.
```python
def data_received(self, data: bytes) -> str:
"""Handle received data.
Args:
data: A bytes-like object (ReactorBuffer) supporting:
- Buffer protocol: memoryview(data) for zero-copy access
- Indexing/slicing: data[0], data[0:10]
- Bytes methods: data.startswith(), data.find(), data.decode()
- Comparison: data == b'...'
- Conversion: bytes(data) creates a copy
Returns:
Action string indicating what to do next
"""
self.request_buffer.extend(data)
if self.request_complete():
self.prepare_response()
return "write_pending"
return "continue" # Need more data
```
#### `write_ready() -> action`
Called when the fd is ready for writing.
```python
def write_ready(self) -> str:
"""Handle write readiness.
Returns:
Action string indicating what to do next
"""
if not self.write_buffer:
return "read_pending"
written = self.write(bytes(self.write_buffer))
del self.write_buffer[:written]
if self.write_buffer:
return "continue" # More to write
return "read_pending" # Done writing
```
#### `connection_lost()`
Called when the connection is closed.
```python
def connection_lost(self):
"""Called when connection closes.
Override to perform cleanup.
"""
# Clean up resources
self.cleanup()
```
### I/O Helper Methods
#### `read(size) -> bytes`
Read from the file descriptor:
```python
data = self.read(65536) # Read up to 64KB
if not data:
return "close" # EOF or error
```
#### `write(data) -> int`
Write to the file descriptor:
```python
written = self.write(response_bytes)
del self.write_buffer[:written] # Remove written bytes
```
## Zero-Copy ReactorBuffer
The `data` argument passed to `data_received()` is a `ReactorBuffer` - a special bytes-like type that provides zero-copy access to read data. The data is read by the NIF before acquiring the GIL, and wrapped in a ReactorBuffer that exposes the memory via Python's buffer protocol.
### Benefits
- **No data copying** - Data goes directly from kernel to Python without intermediate copies
- **Transparent compatibility** - ReactorBuffer acts like `bytes` in all common operations
- **Memory efficiency** - Large payloads don't require extra allocations
### Supported Operations
ReactorBuffer supports all common bytes operations:
```python
def data_received(self, data):
# Buffer protocol - zero-copy access
mv = memoryview(data)
first_byte = mv[0]
# Indexing and slicing
header = data[0:4]
last_byte = data[-1]
# Bytes methods
if data.startswith(b'GET'):
method = 'GET'
pos = data.find(b'\r\n')
count = data.count(b'/')
# String conversion
text = data.decode('utf-8')
# Comparison
if data == b'PING':
self.write_buffer.extend(b'PONG')
# Convert to bytes (creates a copy)
data_copy = bytes(data)
# 'in' operator
if b'HTTP' in data:
self.handle_http()
# Length
size = len(data)
# Extend bytearray (uses buffer protocol)
self.request_buffer.extend(data)
return "continue"
```
### Performance Considerations
The zero-copy benefit is in the NIF read path - data is read directly into a buffer that Python wraps without copying. This avoids the overhead of creating a Python bytes object for every read.
- **NIF read path**: Data goes directly from kernel to Python without intermediate copies
- **Parsing operations**: `startswith()`, `find()` etc. are optimized C implementations
- **Direct memoryview access**: Use `data.memoryview()` for maximum zero-copy performance
- **Creating bytes**: Call `bytes(data)` only when you need a persistent copy
```python
# For maximum performance, use memoryview slicing for comparisons
mv = data.memoryview()
if mv[:3] == b'GET':
# Process GET request
```
## Action Return Values
Protocol methods return action strings that tell Erlang what to do next:
| Action | Description | Erlang Behavior |
|--------|-------------|-----------------|
| `"continue"` | Keep current mode | Re-register same event |
| `"write_pending"` | Ready to write | Switch to write mode (`enif_select` WRITE) |
| `"read_pending"` | Ready to read | Switch to read mode (`enif_select` READ) |
| `"close"` | Close connection | Close fd and call `connection_lost()` |
## Factory Pattern
Register a protocol factory to create protocol instances for each connection:
```python
import erlang.reactor as reactor
class MyProtocol(reactor.Protocol):
# ... implementation
# Set the factory - called for each new connection
reactor.set_protocol_factory(MyProtocol)
# Get the protocol instance for an fd
proto = reactor.get_protocol(fd)
```
## Complete Example: Echo Protocol
Here's a complete echo server protocol:
```python
import erlang.reactor as reactor
class EchoProtocol(reactor.Protocol):
"""Simple echo protocol - sends back whatever it receives."""
def connection_made(self, fd, client_info):
super().connection_made(fd, client_info)
print(f"Connection from {client_info.get('addr')}:{client_info.get('port')}")
def data_received(self, data):
"""Echo received data back to client."""
if not data:
return "close"
# Buffer the data for writing
self.write_buffer.extend(data)
return "write_pending"
def write_ready(self):
"""Write buffered data."""
if not self.write_buffer:
return "read_pending"
written = self.write(bytes(self.write_buffer))
del self.write_buffer[:written]
if self.write_buffer:
return "continue" # More data to write
return "read_pending" # Done, wait for more input
def connection_lost(self):
print(f"Connection closed: fd={self.fd}")
# Register the factory
reactor.set_protocol_factory(EchoProtocol)
```
## Example: HTTP Protocol (Simplified)
```python
import erlang.reactor as reactor
class SimpleHTTPProtocol(reactor.Protocol):
"""Minimal HTTP/1.0 protocol."""
def __init__(self):
super().__init__()
self.request_buffer = bytearray()
def data_received(self, data):
self.request_buffer.extend(data)
# Check for end of headers
if b'\r\n\r\n' in self.request_buffer:
self.handle_request()
return "write_pending"
return "continue"
def handle_request(self):
"""Parse request and prepare response."""
request = self.request_buffer.decode('utf-8', errors='replace')
first_line = request.split('\r\n')[0]
method, path, _ = first_line.split(' ', 2)
# Simple response
body = f"Hello! You requested {path}"
response = (
f"HTTP/1.0 200 OK\r\n"
f"Content-Length: {len(body)}\r\n"
f"Content-Type: text/plain\r\n"
f"\r\n"
f"{body}"
)
self.write_buffer.extend(response.encode())
def write_ready(self):
if not self.write_buffer:
return "close" # HTTP/1.0: close after response
written = self.write(bytes(self.write_buffer))
del self.write_buffer[:written]
if self.write_buffer:
return "continue"
return "close"
reactor.set_protocol_factory(SimpleHTTPProtocol)
```
## Passing Sockets from Erlang to Python
### Method 1: Socket FD Handoff to Reactor
The most efficient way is to hand off the socket's file descriptor directly:
```erlang
%% Erlang: Accept and hand off to Python reactor
{ok, ClientSock} = gen_tcp:accept(ListenSock),
{ok, {Addr, Port}} = inet:peername(ClientSock),
%% Get the raw file descriptor
{ok, Fd} = inet:getfd(ClientSock),
%% Hand off to Python - Erlang no longer owns this socket
py_reactor_context:handoff(Fd, #{
addr => inet:ntoa(Addr),
port => Port,
type => tcp
}).
```
```python
# Python: Protocol handles the fd
import erlang.reactor as reactor
class MyProtocol(reactor.Protocol):
def data_received(self, data):
# self.fd is the socket fd from Erlang
self.write_buffer.extend(b"Got: " + data)
return "write_pending"
reactor.set_protocol_factory(MyProtocol)
```
### Method 2: Pass Socket FD to asyncio
For asyncio-based code, pass the fd and wrap it in Python:
```erlang
%% Erlang: Get fd and pass to Python
{ok, ClientSock} = gen_tcp:accept(ListenSock),
{ok, Fd} = inet:getfd(ClientSock),
%% Call Python with the fd
Ctx = py:context(1),
py:call(Ctx, my_handler, handle_connection, [Fd]).
```
```python
# Python: Wrap fd in asyncio
import asyncio
import socket
async def handle_connection(fd: int):
# Create socket from fd (Python takes ownership)
sock = socket.socket(fileno=fd)
sock.setblocking(False)
# Use asyncio streams
reader, writer = await asyncio.open_connection(sock=sock)
data = await reader.read(1024)
writer.write(b"Echo: " + data)
await writer.drain()
writer.close()
await writer.wait_closed()
def handle_connection_sync(fd: int):
"""Sync wrapper for Erlang call."""
asyncio.run(handle_connection(fd))
```
### Method 3: Pass Socket Object via Pickle (Not Recommended)
For simple cases, you can pickle socket info, but this is less efficient:
```erlang
%% Erlang: Pass connection info
{ok, {Addr, Port}} = inet:peername(ClientSock),
py:call(Ctx, my_handler, connect_to, [Addr, Port]).
```
```python
# Python: Create new connection (less efficient - new socket)
import socket
def connect_to(addr: str, port: int):
sock = socket.create_connection((addr, port))
# ... use socket
```
### Socket Ownership
When passing an fd from Erlang to Python, you must decide who owns it:
**Option 1: Transfer ownership to Python**
Erlang gives up the fd entirely. Don't close the Erlang socket.
```erlang
{ok, ClientSock} = gen_tcp:accept(ListenSock),
{ok, Fd} = inet:getfd(ClientSock),
py_reactor_context:handoff(Fd, #{type => tcp}).
%% Don't close ClientSock - Python owns the fd now
```
**Option 2: Duplicate the fd (recommended)**
Use `py:dup_fd/1` to create an independent copy. Both sides can close their own fd.
```erlang
{ok, ClientSock} = gen_tcp:accept(ListenSock),
{ok, Fd} = inet:getfd(ClientSock),
{ok, DupFd} = py:dup_fd(Fd),
py_reactor_context:handoff(DupFd, #{type => tcp}),
gen_tcp:close(ClientSock). %% Safe - Python has its own fd copy
```
This is safer because:
1. **Erlang controls its socket lifecycle** - GC won't affect Python
2. **Python has its own fd** - Independent of Erlang's socket
3. **No double-close issues** - Each side manages its own fd
## Integration with Erlang
### From Erlang: Starting a Reactor Server
```erlang
%% In your Erlang code
-module(my_server).
-export([start/1]).
start(Port) ->
%% Set up the Python protocol factory first
Ctx = py:context(1),
ok = py:exec(Ctx, <<"
import erlang.reactor as reactor
from my_protocols import MyProtocol
reactor.set_protocol_factory(MyProtocol)
">>),
%% Start accepting connections
{ok, ListenSock} = gen_tcp:listen(Port, [binary, {active, false}, {reuseaddr, true}]),
accept_loop(ListenSock).
accept_loop(ListenSock) ->
{ok, ClientSock} = gen_tcp:accept(ListenSock),
{ok, {Addr, Port}} = inet:peername(ClientSock),
%% Hand off to Python reactor
{ok, Fd} = inet:getfd(ClientSock),
py_reactor_context:handoff(Fd, #{
addr => inet:ntoa(Addr),
port => Port,
type => tcp
}),
accept_loop(ListenSock).
```
### How FDs Are Passed from Erlang to Python
1. Erlang accepts a connection and gets the socket fd
2. Erlang calls `py_reactor_context:handoff(Fd, ClientInfo)`
3. The NIF calls Python's `reactor.init_connection(fd, client_info)`
4. Protocol factory creates a new Protocol instance
5. `enif_select` is registered for read events on the fd
6. When events occur, Python callbacks handle the protocol logic
## Module API Reference
### `set_protocol_factory(factory)`
Set the factory function for creating protocols.
```python
reactor.set_protocol_factory(MyProtocol)
# or with a custom factory
reactor.set_protocol_factory(lambda: MyProtocol(custom_arg))
```
### `get_protocol(fd)`
Get the protocol instance for a file descriptor.
```python
proto = reactor.get_protocol(fd)
if proto:
print(f"Protocol state: {proto.client_info}")
```
### `init_connection(fd, client_info)`
Internal - called by NIF on fd handoff.
### `on_read_ready(fd, data)`
Internal - called by NIF when fd is readable. The `data` argument is a `ReactorBuffer` containing the bytes read from the fd.
### `on_write_ready(fd)`
Internal - called by NIF when fd is writable.
### `close_connection(fd)`
Internal - called by NIF to close connection.
## Subinterpreter Support
The reactor supports isolated subinterpreters via `py_reactor_context`. Each subinterpreter has its own reactor module cache, ensuring protocol factories are isolated between contexts.
```erlang
%% Create context with subinterpreter mode
{ok, Ctx1} = py_reactor_context:start_link(1, subinterp, #{
setup_code => <<"
import erlang.reactor as reactor
reactor.set_protocol_factory(EchoProtocol)
">>
}),
%% Create another context with different protocol
{ok, Ctx2} = py_reactor_context:start_link(2, subinterp, #{
setup_code => <<"
import erlang.reactor as reactor
reactor.set_protocol_factory(HttpProtocol)
">>
}).
```
Each context runs in its own subinterpreter with isolated protocol factory and connection state. This enables running multiple protocol handlers in the same BEAM VM without interference.
## See Also
- [Asyncio](asyncio.md) - Higher-level asyncio event loop for Python
- [Security](security.md) - Security sandbox documentation
- [Getting Started](getting-started.md) - Basic usage guide
