Topic and Pub/Sub

Key Takeaways

After reading this guide, you will understand:

• How Topic provides ultra-fast pub/sub communication (~75ns in-process, ~151-279ns cross-process) with automatic optimization
• The send() and recv() methods for publishing and subscribing to typed topics
• Communication patterns (one-to-one, one-to-many, many-to-one, many-to-many) for different architectures
• When to use Topic for real-time, single-machine communication vs network-based messaging

The Topic is HORUS's ultra-low latency publish-subscribe (pub/sub) communication system. It enables nodes to exchange messages through shared memory IPC with automatic optimization — ~75ns median for a 16-byte message between threads in one process, ~151-279ns between separate processes, depending on topology.

What is a Topic?

A Topic<T> is a typed communication channel that connects publishers and subscribers. The system automatically detects the optimal backend based on:

• How many publishers and subscribers exist
• Whether the message type is simple and fixed-size or needs serialization

You just call send() and recv() — HORUS handles the rest. Participants may share a thread, share a process, or live in different processes; every topic is shared-memory backed either way, so location changes neither the API nor the backend choice.

Key Features

Automatic Optimization: The fastest communication path is auto-detected at runtime

Zero-Copy Communication: Simple fixed-size types are shared directly in memory without serialization

Type Safety: Compile-time guarantees for message types — within one program. Whether a publisher and subscriber in different nodes, processes or languages agree is checked at runtime, when the topic is opened

Live Migration: paths upgrade immediately when a participant joins. A departure is only noticed when that participant's 5-second lease is later reclaimed by a new registration, so a topic can stay on the wider backend for a while after a subscriber exits

Automatic Optimization

HORUS auto-selects the fastest communication path based on your topology. You never need to configure this — it happens automatically when participants call send() or recv().

The message type matters too: simple fixed-size types get a zero-copy ring, everything else gets a serializing one. The rings are internal — there is no backend type to name, import or select. See Performance for measured latency.

If topology changes (e.g., a second subscriber joins), HORUS automatically migrates to the optimal path without dropping messages.

// All of these use the same API — optimization is automatic
let topic: Topic<f32> = Topic::new("velocity")?;
topic.send(1.5);
let msg = topic.recv();

Tip: Simple fixed-size structs (no String, Vec, Box) automatically get a faster zero-copy path. Both types work with the same API — HORUS picks the fastest path for you.

Basic Usage

Creating a Topic

use horus::prelude::*;

// Create a Topic for f32 values on topic "velocity"
let topic: Topic<f32> = Topic::new("velocity")?;

Publishing Messages

use horus::prelude::*;

struct Publisher {
    velocity_pub: Topic<f32>,
}

impl Publisher {
    fn new() -> Result<Self> {
        Ok(Self {
            velocity_pub: Topic::new("velocity")?,
        })
    }
}

impl Node for Publisher {
    fn name(&self) -> &str {
        "Publisher"
    }

    fn tick(&mut self) {
        let velocity = 1.5;

        // Send message — infallible, non-blocking
        self.velocity_pub.send(velocity);
    }
}

Subscribing to Messages

use horus::prelude::*;

struct Subscriber {
    velocity_sub: Topic<f32>,
}

impl Subscriber {
    fn new() -> Result<Self> {
        Ok(Self {
            velocity_sub: Topic::new("velocity")?,
        })
    }
}

impl Node for Subscriber {
    fn name(&self) -> &str {
        "Subscriber"
    }

    fn tick(&mut self) {
        if let Some(velocity) = self.velocity_sub.recv() {
            println!("Received velocity: {}", velocity);
        }
    }
}

Transport

Topics use local shared memory for ultra-fast communication between processes on the same machine.

// Automatically uses local shared memory
let topic: Topic<SensorData> = Topic::new("sensors")?;

• Performance: ~75ns median for a 16-byte message in one process, ~151-279ns across processes
• Use case: All nodes on same machine
• Characteristics: Ultra-fast, deterministic, zero-copy

For multi-machine communication, HORUS supports zero-configuration peer discovery over UDP multicast (group 224.0.69.72, port 9100). Enable it with the net feature flag: horus = { version = "0.4", features = ["net"] }. Override discovery with HORUS_NET_MULTICAST, HORUS_NET_PORT, or HORUS_NET_PEER=ip1,ip2 for direct unicast.

API Reference

Note: HORUS separates introspection from the hot path. The send() and recv() methods have zero logging overhead by default.

send()

pub fn send(&self, msg: T)

Publishes a message to the topic. Fire-and-forget — it returns nothing, so there is no Result to handle, but delivery is not guaranteed.

On a broadcast topology (multiple subscribers) the ring is latest-wins: the oldest slot is overwritten and the new message always lands.

On a point-to-point topology (one subscriber) a full ring first retries — immediately, then a short spin, then a bounded yield. What happens next depends on whether anything is draining the ring:

• A live subscriber is draining it, just slowly. send() gives up and drops the new message. Queued messages keep their FIFO order.
• Nothing is draining it — no subscriber has registered, the subscriber's process is gone, or tail has not moved for a full 5-second lease. send() retires the oldest unread slot and accepts the new message, so the ring holds the most recent capacity messages rather than the first ones. Every retired message is counted in dropped_count().

The second case is the common one for a node run on its own, before anything subscribes. Use try_send() to detect a drop, and read dropped_count() to see how many messages were retired.

// send() returns no Result — but the message can still be dropped
topic.send(data);

recv()

pub fn recv(&self) -> Option<T>

Receives the next message from the topic. Returns None if no message is available. Non-blocking.

if let Some(data) = topic.recv() {
    // Process the received message
}

try_send()

pub fn try_send(&self, msg: T) -> Result<(), T>

Attempts to send a message, returning it on failure. Unlike send(), which on point-to-point topologies retries briefly and then drops the message, try_send() hands it back to you. A broadcast ring is latest-wins and never reports full, so back-pressure is not a failure mode there — but that is not the same as never failing. PodShm is genuinely infallible (send_shm_pod_broadcast has one exit, Ok(())). FanoutShm can still return Err: when all 16 publisher endpoint slots are live, and when the payload does not fit a slot (try_send_serde refuses anything over slot_size). Handle the Err arm on any topology.

match topic.try_send(data) {
    Ok(()) => { /* sent successfully */ }
    Err(returned_data) => { /* ring full, no endpoint slot, or payload too large */ }
}

send_blocking()

pub fn send_blocking(&self, msg: T, timeout: std::time::Duration) -> Result<(), SendBlockingError>

Sends a message, waiting for room in the ring instead of dropping it. It tries once immediately, then spins, then yields, then sleeps in 100µs steps until the deadline.

SendBlockingError has two variants, and the second one is the one to plan for:

SendBlockingError is not in the prelude; import it from horus::communication::SendBlockingError if you need to match on it.

use std::time::Duration;

use horus::communication::SendBlockingError;

match topic.send_blocking(command, Duration::from_millis(5)) {
    Ok(()) => {}
    Err(SendBlockingError::Timeout) => {
        // the ring stayed full for 5ms — escalate rather than continue silently
    }
    Err(SendBlockingError::NoBackpressure) => {
        // this topic is broadcast-backed: nothing was sent, and no timeout will
        // help. Use send() and accept latest-wins, or give the command path its
        // own single-subscriber topic.
    }
}

read_latest()

pub fn read_latest(&self) -> Option<T>
where
    T: Copy,

Returns the most recent message without advancing the consumer position. Calling it multiple times returns the same message until a new one is published. Useful for reading infrequently-updated data like static transforms or configuration.

T: Copy requirement: read_latest() requires T: Copy to guarantee safe concurrent reads. Types with heap allocations (String, Vec, etc.) should use recv() instead.

// Read latest transform (doesn't consume it)
if let Some(transform) = tf_topic.read_latest() {
    self.apply_transform(transform);
}

has_message()

pub fn has_message(&self) -> bool

Checks if at least one message is available without consuming it.

pending_count()

pub fn pending_count(&self) -> u64

Returns the number of messages waiting to be consumed.

missed_count()

pub fn missed_count(&self) -> u64

Returns how many messages this subscriber was lapped past by the producer. It is counted per handle on the consumer side, where the loss happens, so two subscribers on the same topic report independently. This is the counter that detects a slow consumer — dropped_count() stays at 0 on a broadcast topic, because an overwriting producer never fails a send.

provides_backpressure()

pub fn provides_backpressure(&self) -> bool

Whether a full ring can make this topic's try_send refuse a message. It is false on the broadcast backends — which is what a topic silently becomes once a second subscriber attaches — and therefore false wherever send_blocking() will return NoBackpressure.

Call it after the first send() or recv(). A handle does not resolve its backend until it moves a message, and an unresolved backend also answers false, so an assertion placed before the first send fires on every topic including the ones that are fine.

estop.send(command);   // resolves the backend
assert!(
    estop.provides_backpressure(),
    "e-stop settled on a lossy backend"
);

Runtime Debug Logging

Debug logging is toggled at runtime from the TUI monitor — no code changes or recompilation needed. Select a topic in the Topics tab and press Enter to start logging; press Esc to stop.

When debug logging is active, every send() and recv() records a metadata entry — topic name, direction, and IPC latency — to the log buffer. The Rust transport never calls log_summary(), so implementing LogSummary does not change what a topic logs; use it yourself inside hlog! when you want the message contents in the log.

horus monitor --tui    # Open TUI, navigate to Topics tab, press Enter on a topic

Debug logging adds no overhead when disabled — introspection is fully separated from the hot path.

with_capacity()

pub fn with_capacity(name: &str, capacity: u32, slot_size: Option<usize>) -> Result<Self>

Creates a topic with a custom ring buffer capacity. By default, HORUS auto-sizes capacity based on the message type size. Use this when you need more buffering (e.g., bursty producers).

// 256-slot ring buffer (rounded up to next power of 2)
let topic: Topic<f32> = Topic::with_capacity("velocity", 256, None)?;

Type-Safe Topic Descriptors

The topics! macro defines compile-time topic descriptors that prevent topic name typos and keep topic names consistent across your codebase.

Defining Topics

use horus::prelude::*;

// `topics!` is not in the prelude — invoke it through the re-exported crate
horus::topics! {
    pub CMD_VEL: CmdVel = "cmd_vel",
    pub SENSOR_DATA: SensorReading = "sensor.data",
    pub MOTOR_STATUS: MotorState = "motor.status",
}

Using Descriptors

// Publisher — the descriptor supplies the name; the message type is chosen here
let pub_topic: Topic<CmdVel> = Topic::new(CMD_VEL.name())?;
pub_topic.send(CmdVel::new(1.0, 0.5));

// Subscriber — the descriptor supplies the name; the message type is chosen here
let sub_topic: Topic<CmdVel> = Topic::new(CMD_VEL.name())?;
if let Some(cmd) = sub_topic.recv() {
    // cmd is guaranteed to be CmdVel
}

This prevents common errors like:

• Typos in topic names (caught at compile time)
• Inconsistent topic names between publisher and subscriber, or across modules
• Topic names drifting out of sync when one call site is renamed

The declared message type in topics! is documentation only — TopicDescriptor<T> carries T as PhantomData and name() returns a plain &'static str, so each call site still picks its own Topic<T>.

Communication Patterns

The API is identical in every pattern — only the ring HORUS picks underneath changes. Measured latency for each topology lives in Performance below.

One-to-One

Single publisher, single subscriber — a point-to-point ring, FIFO.

struct PubNode {
    data_pub: Topic<f32>,
}

impl Node for PubNode {
    fn name(&self) -> &str { "PubNode" }

    fn tick(&mut self) {
        self.data_pub.send(42.0);
    }
}

struct SubNode {
    data_sub: Topic<f32>,
}

impl Node for SubNode {
    fn name(&self) -> &str { "SubNode" }

    fn tick(&mut self) {
        if let Some(data) = self.data_sub.recv() {
            println!("Got: {}", data);
        }
    }
}

One-to-Many (Broadcast)

Single publisher, multiple subscribers — a broadcast ring, so each subscriber reads the stream independently rather than competing for messages. One slow subscriber does not starve the others.

That is independence, not a delivery guarantee. The ring is latest-wins: a publisher that laps a subscriber overwrites slots that subscriber has not read, and the subscriber jumps forward to about half a ring back from the head — not to the newest message — and the skipped messages are lost. Landing next to the write cursor would put the consumer on the slot the producer overwrites next, so it would lap again on the following poll.

The skip raises no error, but it is counted: missed_count() returns how many messages this subscriber was lapped past, tallied per handle on the consumer side. What stays at 0 is dropped_count() — it reports send-side failures only, and an overwriting producer never fails a send.

send_blocking() does not close the gap: on a broadcast ring it returns Err(SendBlockingError::NoBackpressure) without sending, so it turns lost messages into no message at all. What works is keeping the subscriber fast enough to drain the ring, giving the topic more capacity with Topic::with_capacity, or moving the slow work off the subscriber's tick.

// One publisher
struct Broadcaster {
    broadcast_pub: Topic<String>,
}

// Multiple subscribers — each reads the stream independently
struct Listener1 { broadcast_sub: Topic<String> }
struct Listener2 { broadcast_sub: Topic<String> }
struct Listener3 { broadcast_sub: Topic<String> }

Many-to-One (Aggregation)

Multiple publishers, single subscriber — a multi-producer point-to-point ring, FIFO across all publishers.

// Multiple publishers
struct Sensor1 { reading_pub: Topic<f32> }
struct Sensor2 { reading_pub: Topic<f32> }

// Single aggregator
struct Aggregator {
    reading_sub: Topic<f32>,
}

impl Node for Aggregator {
    fn name(&self) -> &str { "Aggregator" }

    fn tick(&mut self) {
        if let Some(reading) = self.reading_sub.recv() {
            self.process(reading);
        }
    }
}

Many-to-Many

Multiple publishers and subscribers — a broadcast ring, same as one-to-many.

struct Agent1 {
    state_pub: Topic<RobotState>,
    state_sub: Topic<RobotState>,
}

struct Agent2 {
    state_pub: Topic<RobotState>,
    state_sub: Topic<RobotState>,
}

Topic Naming

Use Dots, Not Slashes

Important: HORUS uses dots (.) for topic hierarchy, not slashes (/).

// CORRECT - Use dots
let topic = Topic::new("sensors.lidar");
let topic = Topic::new("robot.cmd_vel");

// DISCOURAGED - slashes work, but break the convention
let topic = Topic::new("sensors/lidar");   // Discouraged — works, but breaks the dot convention
let topic = Topic::new("robot/cmd_vel");   // Discouraged — works, but breaks the dot convention

Why dots instead of slashes?

Dots are the HORUS convention; slashes are accepted but produce nested directories under /dev/shm/horus_<ns>/topics/. A leading slash (ROS-style /sensor/lidar) is rejected as an absolute path.

Coming from ROS?

Best Practices

// Descriptive names
let topic = Topic::new("cmd_vel");           // Good
let topic = Topic::new("data");              // Too vague

// Hierarchical naming
let topic = Topic::new("sensor.lidar");      // Hierarchical
let topic = Topic::new("robot1.cmd_vel");    // Namespaced
let topic = Topic::new("diagnostics.cpu");   // Categorized

Reserved Topic Names

Avoid using these patterns:

• Topics starting with _ (internal use)
• Topic names beginning with / (rejected as an absolute path)
• Topics with special characters: !@#$%^&*()

Error Handling

Send Behavior

send() is fire-and-forget — it returns nothing, so there is no Result to handle, but delivery is not guaranteed. On broadcast topologies (multiple subscribers) the ring is latest-wins: the oldest slot is overwritten. On point-to-point topologies a full ring retries briefly, then either drops the new message (if a live subscriber is draining) or retires the oldest unread slot to make room (if nothing is draining it — including the case where nothing has subscribed yet). See send() above for the full rule.

// send() returns no Result — but the message can still be dropped
topic.send(data);

For explicit error handling, use try_send():

match topic.try_send(data) {
    Ok(()) => { /* success */ }
    Err(msg) => { /* buffer full — msg returned to caller */ }
}

Receive Behavior

recv() returns None when no message is available — this is normal, not an error:

match topic.recv() {
    Some(data) => {
        // Process data
    }
    None => {
        // No data available - this is normal
    }
}

Message Types

What Types Can I Use?

Most types work with Topic — just add standard derives:

use serde::{Serialize, Deserialize};

// Primitive types work out of the box
let topic: Topic<f32> = Topic::new("float_topic")?;
let topic: Topic<bool> = Topic::new("bool_topic")?;

// Custom structs — just add derives
#[derive(Clone, Serialize, Deserialize)]
struct MyMessage {
    x: f32,
    y: f32,
    name: String,
}

let topic: Topic<MyMessage> = Topic::new("my_topic")?;

// Collections work too
#[derive(Clone, Serialize, Deserialize)]
struct SensorBatch {
    readings: Vec<f32>,
    timestamp: u64,
}

let topic: Topic<SensorBatch> = Topic::new("batch_topic")?;

Required Traits

Your message types need these traits (all auto-derived):

Additionally, types must be Send + Sync + 'static — satisfied automatically by most types.

use serde::{Serialize, Deserialize};

#[derive(Clone, Serialize, Deserialize)]
struct MyMessage {
    // your fields
}

Advanced Usage

Conditional Publishing

Only publish when certain conditions are met:

impl Node for ConditionalPublisher {
    fn tick(&mut self) {
        let data = self.read_sensor();

        if data > self.threshold {
            self.alert_pub.send(data);
        }
    }
}

Message Buffering

Cache the last received message using read_latest() or manual buffering:

struct BufferedSubscriber {
    data_sub: Topic<f32>,
    last_value: Option<f32>,
}

impl Node for BufferedSubscriber {
    fn tick(&mut self) {
        if let Some(value) = self.data_sub.recv() {
            self.last_value = Some(value);
        }

        if let Some(value) = self.last_value {
            self.process(value);
        }
    }
}

For Copy types, use read_latest() instead — it reads the latest message without consuming it:

// Requires T: Copy — reads latest without advancing consumer position
if let Some(transform) = self.tf_sub.read_latest() {
    self.apply_transform(transform);
}

Rate Limiting

Publish at a specific rate:

struct RateLimitedPublisher {
    data_pub: Topic<f32>,
    tick_count: u32,
    publish_every_n_ticks: u32,
}

impl Node for RateLimitedPublisher {
    fn tick(&mut self) {
        self.tick_count += 1;

        if self.tick_count % self.publish_every_n_ticks == 0 {
            self.data_pub.send(42.0);
        }
    }
}

Message Filtering

Filter messages before processing:

impl Node for FilteringSubscriber {
    fn tick(&mut self) {
        if let Some(data) = self.data_sub.recv() {
            if data.is_valid() && data.quality > 0.8 {
                self.process(data);
            }
        }
    }
}

Memory and Capacity

Memory Usage Per Topic

By default, Topic::new() auto-sizes capacity based on message type size:

Auto capacity is always clamped to [16, 1024] and rounded up to the next power of two. A fixed-size (POD) type of 56 bytes or less whose alignment is 8 or less gets a padded 64-byte slot — one cache line holding an 8-byte readiness stamp next to the payload. A POD type that is larger, or aligned above 8 (a u128 field, #[repr(align(16))], an SSE vector), gets a slot exactly size_of::<T>() wide instead: the co-located payload always starts 8 bytes into a cache-line-aligned slot, and nothing rounds that up, so a 16-aligned type placed there would be written through a misaligned vector store. Only non-POD (heap-owning) types fall back to the 8 KB default slot.

The region is a 640-byte header plus the slots. A co-located slot carries its own stamp, so the first four rows above are 640 + capacity × 64 and nothing more; the split layouts — anything over 56 bytes, and anything aligned above 8 — allocate a separate 8-byte sequence entry per slot on top of the data, which is why Imu at 16 × 304 bytes still comes to ~5.5 KB.

For most robotics applications, memory usage per topic is under 1 MB.

Cleaning Up

A topic's shared memory file survives as long as any process still has it mapped, so a publisher can exit and restart without taking the ring away from its subscribers. What outlives a whole run is the session namespace directory and the non-topic files under it. Clean up between sessions:

# Clean shared memory only (recommended)
horus clean --shm

# Preview what would be cleaned
horus clean --shm --dry-run

# Clean everything (shared memory + build cache)
horus clean --all

The last process to release a topic unlinks its file. Every holder keeps a shared flock on the backing file for the life of its mapping, and the region's drop takes a non-blocking exclusive lock as a last-one-out test: if it succeeds, the file is removed. Only a crash — where drop never ran — leaves a topic file behind, and that one is reused, not deleted: its header is re-initialised when a live process next opens the same topic.

What accumulates in /dev/shm across runs is the per-session namespace directory and the artifacts under it that nothing unlinks — fanout locks, log files, tensor pool files. horus clean --shm clears those.

Troubleshooting

"No space left on device" — Shared memory is full:

horus clean --shm            # Clean up

Type mismatch — Ensure publisher and subscriber use the exact same type for a topic name:

// Both sides MUST use the same type
let pub_topic: Topic<CmdVel> = Topic::new("cmd_vel")?;
let sub_topic: Topic<CmdVel> = Topic::new("cmd_vel")?;

Performance

Latency by Topology (16B message)

Every topic is shared-memory backed, so these are end-to-end one-way IPC figures, not in-process function calls. Measured by all_paths_latency on an Intel Core i7-10750H (6C/12T, powersave governor), 100K samples per scenario:

The ~79ns row is the bare atomic round of a shared-memory handoff with no framework on top — a bound HORUS approaches, not a latency any topic achieves.

Latency by Message Type

Per-message-type figures come from robotics_messages_benchmark (same process, publisher and subscriber on separate threads, 50K iterations). CmdVel (16B) is the reference point at ~75ns median / ~135ns p99; see Benchmarks for the other message types.

Latency grows with message size, with one exception: Image, PointCloud, DepthImage, OccupancyGrid and CostMap are pool-backed — only a small fixed-size descriptor (224-440 bytes, depending on the type) crosses the ring, so their latency is independent of payload size.

Throughput

• Millions of messages per second for small messages
• Gigabytes per second for large messages
• Deterministic latency regardless of system load

Best Practices

Use simple fixed-size types when possible — they get the fastest path automatically:

Topic::<[f32; 3]>::new("position")?;   // Fixed-size: fastest path
Topic::<Vec<f32>>::new("position")?;   // Dynamic: still fast, but uses serialization

Keep messages small — latency grows with size.

Check recv() every tick — don't skip ticks:

fn tick(&mut self) {
    if let Some(msg) = self.sub.recv() {
        self.process(msg);
    }
}

Use topics! macro for shared topic definitions across modules:

horus::topics! {
    pub CMD_VEL: CmdVel = "cmd_vel",
    pub ODOM: Odometry = "odom",
}

send() returns no Result — but delivery is not guaranteed, so reach for try_send() or send_blocking() on critical topics:

self.telemetry_pub.send(sample);              // best-effort — may be dropped under load

if self.cmd_pub.try_send(command).is_err() {
    // ring full — the command was returned, handle or retry it
}

Next Steps

• Learn about the Scheduler for orchestrating nodes
• Explore Message Types for standard robotics messages
• Read the API Reference for complete Topic documentation