# Message Passing Using Cereal Messaging in Openpilot: Implementation Strategy and Architecture

> Discover Openpilot's dual-layer message passing strategy using Cereal for typed serialization and msgq for zero-copy transport. Learn about PubMaster and SubMaster architecture.

- Repository: [comma.ai/openpilot](https://github.com/commaai/openpilot)
- Tags: architecture
- Published: 2026-03-05

---

**Openpilot implements message passing using a dual-layer architecture where Cereal (Cap'n Proto schemas) provides strongly-typed serialization and msgq provides zero-copy shared-memory transport, orchestrated by PubMaster and SubMaster classes that abstract socket lifecycle and frequency tracking.**

Openpilot’s real-time autonomous driving stack requires high-throughput, low-latency communication between perception, planning, and control processes. The commaai/openpilot repository solves this with a specialized **message passing using cereal messaging** layer that combines Cap'n Proto schemas for type safety with a lightweight shared-memory queue system. This design eliminates serialization overhead while maintaining strict data contracts across Python and C++ boundaries.

## The Two-Layer Architecture: Cereal and msgq

The implementation separates message typing from transport mechanics. **Cereal** defines all data structures in `cereal/log.capnp` (compiled to [`cereal/log.py`](https://github.com/commaai/openpilot/blob/main/cereal/log.py)), generating typed Cap'n Proto objects. **msgq** provides the underlying shared-memory transport, acting as a ZeroMQ-like wrapper that bypasses kernel networking stacks for intra-process communication.

This separation allows publishers to construct strongly-typed messages while the transport layer handles raw bytes through lock-free shared memory segments. The integration point resides in [`cereal/messaging/__init__.py`](https://github.com/commaai/openpilot/blob/main/cereal/messaging/__init__.py), where high-level Python classes wrap the raw msgq API.

## Service Registry and Queue Configuration

All message types and their transport parameters are centralized in [`cereal/services.py`](https://github.com/commaai/openpilot/blob/main/cereal/services.py). The `SERVICE_LIST` dictionary (constructed from the `_services` tuple at lines [22‑27](https://github.com/commaai/openpilot/blob/master/cereal/services.py#L22-L27)) defines every service’s logging policy, frequency, and queue size:

```python
_services = {
    "gyroscope": (True, 104., 104),            # (should_log, frequency, decimation)

    "controlsState": (True, 100., 10, QueueSize.MEDIUM),
    # … additional services …

}
SERVICE_LIST = {name: Service(*vals) for idx, (name, vals) in enumerate(_services.items())}

```

Each entry specifies a **queue size** (big, medium, or small) that determines the shared-memory segment capacity. High-rate streams like `can` or `controlsState` receive larger buffers, while low-rate diagnostic streams remain lightweight.

## Socket Initialization and Shared-Memory Segments

The [`cereal/messaging/__init__.py`](https://github.com/commaai/openpilot/blob/main/cereal/messaging/__init__.py) module provides factory functions that consult the service registry to size transport segments appropriately. The `pub_sock` function (lines [19‑23](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L19-L23)) creates publisher sockets:

```python
def pub_sock(endpoint: str) -> PubSocket:
    service = SERVICE_LIST.get(endpoint)
    segment_size = service.queue_size if service else 0
    return msgq.pub_sock(endpoint, segment_size)

```

Similarly, `sub_sock` (lines [25‑31](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L25-L31)) initializes subscriber sockets with optional polling and conflation:

```python
def sub_sock(endpoint: str, poller: Optional[Poller] = None,
             addr: str = "127.0.0.1", conflate: bool = False,
             timeout: Optional[int] = None) -> SubSocket:
    service = SERVICE_LIST.get(endpoint)
    segment_size = service.queue_size if service else 0
    return msgq.sub_sock(endpoint, poller=poller, addr=addr,
                         conflate=conflate, timeout=timeout,
                         segment_size=segment_size)

```

The **segment size** parameter ensures that high-frequency message streams allocate sufficient shared memory to prevent dropped frames during burst transfers.

## Publishing Messages with PubMaster

The `PubMaster` class (lines [50‑66](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L50-L66)) provides a high-level interface for emitting messages. It maintains a dictionary of publisher sockets created via `pub_sock`:

```python
class PubMaster:
    def __init__(self, services: List[str]):
        self.sock = {}
        for s in services:
            self.sock[s] = pub_sock(s)

    def send(self, s: str, dat: Union[bytes, capnp.lib.capnp._DynamicStructBuilder]) -> None:
        if not isinstance(dat, bytes):
            dat = dat.to_bytes()
        self.sock[s].send(dat)

```

Publishers construct messages using the `new_message` helper (lines [42‑55](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L42-L55)), which initializes a `log.Event` structure with common metadata:

```python
def new_message(service: Optional[str], size: Optional[int] = None, **kwargs) -> capnp.lib.capnp._DynamicStructBuilder:
    args = {
        'valid': False,
        'logMonoTime': int(time.monotonic() * 1e9),
        **kwargs
    }
    dat = log.Event.new_message(**args)
    if service is not None:
        if size is None:
            dat.init(service)
        else:
            dat.init(service, size)
    return dat

```

This helper automatically timestamps messages and initializes the service-specific sub-message within the root Cap'n Proto structure.

## Subscribing to Topics with SubMaster

The `SubMaster` class (lines [50‑98](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L50-L98)) manages subscription lifecycle and message bookkeeping. During initialization, it creates `SubSocket` instances for each service, applying `conflate=True` to drop intermediate frames and retain only the latest message:

```python
class SubMaster:
    def __init__(self, services: List[str], poll: Optional[str] = None, ...):
        self.poller = Poller()
        polled_services = set([poll] if poll else services)
        self.non_polled_services = set(services) - polled_services

        for s in services:
            p = self.poller if s not in self.non_polled_services else None
            self.sock[s] = sub_sock(s, poller=p, conflate=True)

            # Pre-allocate empty message for fast reads

            try:
                data = new_message(s)
            except capnp.lib.capnp.KjException:
                data = new_message(s, 0)
            self.data[s] = getattr(data.as_reader(), s)

```

The `update()` method polls registered sockets and converts raw bytes to typed objects using `log_from_bytes` (lines [37‑40](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L37-L40)). Stream health monitoring is handled by `FrequencyTracker` (lines [101‑124](https://github.com/commaai/openpilot/blob/master/cereal/messaging/__init__.py#L101-L124)), which validates that messages arrive at expected frequencies.

## Practical Implementation Examples

**Publishing a controls state message:**

```python
from cereal.messaging import PubMaster, new_message

pub = PubMaster(['controlsState'])

# Build and populate message

msg = new_message('controlsState')
cs = msg.controlsState
cs.active = True
cs.vEgo = 15.2
cs.steeringAngleDeg = 0.3
cs.valid = True

# Transmit via shared memory

pub.send('controlsState', msg)

```

**Subscribing to multiple sensor streams:**

```python
from cereal.messaging import SubMaster

sub = SubMaster(['controlsState', 'carState'], poll='controlsState')

while True:
    sub.update()  # Polls sockets and deserializes data

    if sub.all_alive():
        cs = sub['controlsState']
        car = sub['carState']
        print(f"Speed: {cs.vEgo:.1f} m/s")

```

## Summary

- **Cereal** provides Cap'n Proto schemas in `cereal/log.capnp` for type-safe message definitions that work across Python and C++.
- **Service registry** in [`cereal/services.py`](https://github.com/commaai/openpilot/blob/main/cereal/services.py) (lines 22‑27) automatically configures queue sizes and frequencies for each message type.
- **msgq** enables zero-copy transport through shared-memory segments sized according to service requirements.
- **PubMaster** (lines 50‑66) and **SubMaster** (lines 50‑98) abstract socket lifecycle, serialization via `to_bytes()` and `log_from_bytes`, and frequency tracking.
- The **conflate** mechanism in `sub_sock` (lines 25‑31) ensures subscribers always read the latest frame, critical for real-time control loops.

## Frequently Asked Questions

### How does Cereal messaging differ from standard ROS or MQTT pub/sub?

Unlike ROS or MQTT, which typically serialize to byte streams over TCP/IP, Openpilot's **message passing using cereal messaging** employs Cap'n Proto for serialization combined with msgq for zero-copy shared-memory transport. This eliminates network stack latency and serialization overhead, achieving microsecond-level inter-process communication suitable for real-time vehicle control.

### What determines the buffer size for a message topic?

The `SERVICE_LIST` dictionary in [`cereal/services.py`](https://github.com/commaai/openpilot/blob/main/cereal/services.py) defines the **queue size** (big, medium, or small) for each service based on its frequency and data volume. The `pub_sock` and `sub_sock` functions (lines 19‑31) translate these enum values into specific segment sizes passed to msgq, ensuring high-rate streams like `can` or `controlsState` receive sufficiently large shared-memory buffers.

### Can publishers and subscribers be written in different languages?

Yes. While the high-level `PubMaster` and `SubMaster` classes are Python wrappers, the underlying msgq transport and Cap'n Proto schemas in `cereal/log.capnp` have C++ implementations. The same shared-memory segments and serialization format are accessible from both languages, enabling mixed-language pipelines where C++ modules publish sensor data consumed by Python planning algorithms.

### How does SubMaster handle message frequency validation?

The `SubMaster` class incorporates `FrequencyTracker` (lines 101‑124) to monitor each service's arrival rate against the expected frequency defined in the service registry. The `all_alive()` and `alive` properties check whether messages arrive within tolerance windows, allowing downstream consumers to detect sensor failures or timing violations in the **cereal messaging** pipeline.