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

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), 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, 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. The SERVICE_LIST dictionary (constructed from the _services tuple at lines 22‑27) defines every service’s logging policy, frequency, and queue size:

_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 module provides factory functions that consult the service registry to size transport segments appropriately. The pub_sock function (lines 19‑23) creates publisher sockets:

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) initializes subscriber sockets with optional polling and conflation:

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) provides a high-level interface for emitting messages. It maintains a dictionary of publisher sockets created via pub_sock:

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), which initializes a log.Event structure with common metadata:

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) 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:

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). Stream health monitoring is handled by FrequencyTracker (lines 101‑124), which validates that messages arrive at expected frequencies.

Practical Implementation Examples

Publishing a controls state message:

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:

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 (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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}