Msg: C++20 Shared-Memory & Network Pub/Sub

Distributed systems, high-performance computing clusters, robotics platforms, and real-time simulations rarely rely on a single transport. Local processes may exchange large I/Q sample blocks through shared memory, while remote nodes use UDP multicast, NATS, Eclipse Zenoh, or Redis.

Msg is a modern C++20 library designed to decouple application code from transport mechanics. It unifies local IPC, network transports, in-flight compression, and message serialization behind a URI-driven publish/subscribe interface. The shared-memory backend also provides a standalone API for writing directly into shared buffers without a transport payload copy.

Updated October 10, 2026 with shared-memory streaming, current benchmark results, and build and platform support.

URI-Based Configuration

Rather than writing transport-specific initialization boilerplate, msg configures endpoints via a concise URI schema:

<transport>;<compression>;<address>

or simplified 2-part format (defaulting publishers to none compression):

<transport>;<address>

Supported Transports

Shared memory uses a local channel name, such as shm;none;iq-stream?slots=8,size=4MiB,timeout=1s. Zenoh accepts endpoints such as tcp://127.0.0.1:7447 and tcp://localhost:7447, or zenoh;none;peer for peer discovery.

Transparent 2-Bit Compression Engine

In traditional messaging libraries, publishers and subscribers must agree in advance on compression algorithms. msg features a transparent sender-driven compression model:

• 1-Byte Message Header: Each outgoing message prepends a lightweight header containing a 2-bit compression identifier:
  • 0b00 (none): Uncompressed data (passed as a payload view without a decompression copy).
  • 0b01 (lz4): LZ4 compression for high-bandwidth real-time streams.
  • 0b10 (zstd / zstd:<level>): High-ratio Facebook Zstandard compression (e.g. zstd:5 or zstd:20) for bandwidth-constrained links.
  • 0b11: Reserved for future expansion.
• Dynamic Subscriber Decompression: Subscribers inspect the 2-bit header per frame and call the appropriate static decompression function, without virtual dispatch to a decompressor. Compressed messages allocate a decoded output buffer; LZ4 and Zstandard frames also carry a four-byte original-size field.
• Heterogeneous Streams: A subscriber (e.g., initialized with zenoh;tcp://127.0.0.1:7447) can simultaneously receive from multiple publishers sending with none, lz4, and zstd on the same topic without any configuration changes.

This framing applies to the URI-based API. The standalone ShmPub/ShmSub API carries raw bytes and bypasses compression and serialization.

Shared-Memory Streaming

The shm transport supports up to 64 publishers and 64 subscribers per channel/topic, across processes on the same host. All participants share one ring, and a writer gate serializes publication so every subscriber sees the same merged message order. Adding readers does not create another payload ring, though each reader has its own cursor, semaphore, and worker thread.

The standard msg::Pub/msg::Sub, JSON, and FlatBuffers interfaces support SHM in a normal build. A channel address accepts comma-separated options:

Publishers and subscribers must use the same channel, exact topic, slot count, and slot size. Total mapping size is limited to 1 GiB. Size and timeout units are required and case-sensitive: 4MiB and 1000ms are valid. Omit the options to use the defaults, for example shm;none;iq-stream.

Direct Writes and Buffer Ownership

For the lowest overhead, include <msg/shm.hpp> and link msg::shm. send and try_send copy the payload once into the ring. try_publish(size, writer) lets the application fill a shared slot directly, while the subscriber borrows that slot until its callback returns:

#include <msg/shm.hpp>
#include <chrono>
#include <future>

int main() {
    std::promise<std::size_t> received;
    auto completed = received.get_future();

    msg::ShmSub sub("iq-stream", "iq", [&](std::span<const uint8_t> samples) {
        // Consume here; copy samples if they must outlive this callback.
        received.set_value(samples.size());
    });
    msg::ShmPub pub("iq-stream", "iq");

    const bool sent = pub.try_publish(65536, [](std::span<uint8_t> destination) {
        // Replace this pattern with packed I/Q generation into the shared slot.
        for (std::size_t i = 0; i < destination.size(); ++i) {
            destination[i] = static_cast<uint8_t>(i);
        }
    });
    if (!sent) {
        return 1; // Ring full or another writer active; retry or report a drop.
    }
    return completed.wait_for(std::chrono::seconds(2)) == std::future_status::ready
        ? 0 : 1;
}

Slow subscribers apply backpressure: a slot stays protected until every reader consuming it has finished. try_send and try_publish return false on congestion or writer contention; blocking send retries up to its timeout and then throws. New subscribers receive subsequent messages without replay, and sending with no subscribers succeeds but discards the message.

Reader callbacks and direct-write spans cannot retain their borrowed buffers. Direct-write callbacks run under the writer gate, so keep them short and finish all writes before returning. This removes transport copies between CPU buffers; GPU readback, upload, and downstream processing still have their own costs.

The backend reclaims dead process registrations and recovers from a publisher dying mid-write. Idle readers wait on native semaphores. The current version-3 ring layout supports 64 readers and cannot attach to older layouts; stop old participants cleanly or use a fresh channel name when upgrading. See the shared-memory guide for ownership, recovery, and cleanup details.

Code Examples

Keep subscribers alive while receiving. The network examples assume the relevant broker or Zenoh connection is ready; pub/sub does not replay messages sent before a subscriber connects.

1. Zenoh Pub/Sub with Transparent LZ4 Compression

Publishing with LZ4 compression while the subscriber receives transparently:

#include "msg/msg.hpp"
#include <chrono>
#include <future>
#include <iostream>
#include <span>
#include <string>
#include <string_view>
#include <thread>

int main() {
    const std::string endpoint = "tcp://127.0.0.1:7447";
    std::promise<void> received;
    auto completed = received.get_future();

    // Subscriber does not need to specify compression:
    // It dynamically inspects the 2-bit header and decompresses incoming frames!
    msg::Sub sub("zenoh;" + endpoint, "telemetry", [&](std::span<const uint8_t> payload) {
        std::string_view message(reinterpret_cast<const char*>(payload.data()), payload.size());
        std::cout << "Received telemetry: " << message << "\n";
        received.set_value();
    });

    // Publisher configures LZ4 compression in-flight
    msg::Pub pub("zenoh;lz4;" + endpoint, "telemetry");
    // Allow discovery and subscription propagation in this local demo.
    std::this_thread::sleep_for(std::chrono::seconds(1));

    std::string payload = "SensorPacket: lat=38.2910 lon=-76.5421 alt=1200.5";
    pub.send({reinterpret_cast<const uint8_t*>(payload.data()), payload.size()});

    return completed.wait_for(std::chrono::seconds(2)) == std::future_status::ready
        ? 0 : 1;
}

2. Strongly-Typed Google FlatBuffers

msg integrates with Google FlatBuffers for typed publishing and direct reads of the received serialized buffer. Publishing packs a native object into a FlatBuffer, and the wrappers use the schema’s fully qualified type name as the topic. This example uses the repository’s msg_tests::Person schema, generated with the object API and type names enabled.

#include "msg/msg.hpp"
#include "person_generated.h"
#include <chrono>
#include <future>
#include <iostream>

using msg_tests::Person;
using msg_tests::PersonT;

int main() {
    std::promise<void> received;
    auto completed = received.get_future();

    // Strongly-typed FlatBuffer subscriber
    msg::FbsSub<Person> sub("nats;nats://127.0.0.1:4222", [&](const Person& person) {
        std::cout << "Received Person ID: " << person.id()
                  << ", Name: " << person.name()->string_view() << "\n";
        received.set_value();
    });

    // Strongly-typed FlatBuffer publisher with Zstandard compression
    msg::FbsPub<Person> pub("nats;zstd:10;nats://127.0.0.1:4222");

    PersonT person;
    person.id = 12893;
    person.name = "Alice";
    person.email = "alice@example.com";

    pub.send(person);

    return completed.wait_for(std::chrono::seconds(2)) == std::future_status::ready
        ? 0 : 1;
}

3. JSON Pub/Sub

#include "msg/msg.hpp"
#include "nlohmann/json.hpp"
#include <chrono>
#include <future>
#include <iostream>

int main() {
    std::promise<void> received;
    auto completed = received.get_future();

    msg::JsonSub sub("redis;127.0.0.1:6379", "sensors", [&](const nlohmann::json& data) {
        std::cout << "Sensor ID: " << data["id"] << ", Value: " << data["value"] << "\n";
        received.set_value();
    });

    msg::JsonPub pub("redis;127.0.0.1:6379", "sensors");
    pub.send({{"id", 42}, {"value", 98.6}});

    return completed.wait_for(std::chrono::seconds(2)) == std::future_status::ready
        ? 0 : 1;
}

Shared-Memory Benchmarks

The October 9, 2026 measurements used an AMD Ryzen 9 7950X on Linux, GCC 16.2.1, a Release build with MSG_MAX_PERFORMANCE=OFF, one publisher and one subscriber in separate processes, and eight ring slots. Each case delivered 32 GiB. The table shows medians from three runs with sample validation, which checks message sequence, size, and the final payload byte.

Separate runs checking every payload byte after the sequence/timestamp metadata delivered 5.28–5.43 GB/s across those sizes and modes, limited by reader validation work. All runs delivered the expected message counts and order. Latency is measured from after payload generation to the reader callback; copy-mode latency includes the copy into the ring.

These are CPU transport measurements using reusable buffers, with cache effects and no GPU readback or waterfall processing. They are not sustained DRAM bandwidth or complete application throughput measurements, and the single-reader rates do not describe 64-reader fanout. The recorded results include individual runs and validation settings.

The standalone benchmark runner supports multiple publisher and subscriber processes, copy and direct-write modes, and JSON output with throughput and p50/p99 latency:

python src/msg/bench/run_shm_bench.py build-shm/src/msg/shm/msg_shm_bench \
    --bytes 1073741824 --publishers 1 4 --subscribers 1 16 64 \
    --validation full --json shm-results.json

The normal-build msg_bench_sender and msg_bench_receiver also accept SHM URIs.

Building and Platform Support

The standalone shared-memory target needs C++20 and CMake 3.28+, without third-party messaging, compression, or serialization dependencies:

cmake -S . -B build-shm -DMSG_SHM_ONLY=ON -DCMAKE_BUILD_TYPE=Release
cmake --build build-shm --config Release --parallel
ctest --test-dir build-shm -C Release --output-on-failure

After installation, consumers can use find_package(msg CONFIG REQUIRED) and link msg::shm. The URI-based API and serialization wrappers require a normal build. Python 3 enables the standalone process tests and benchmark runner.

Normal builds manage dependencies through CPM.cmake, or locate installed packages with -DMSG_USE_SYSTEM_DEPS=ON (the default when using vcpkg). Zenoh is enabled by default when cargo is found or system dependencies are selected, and can be controlled with MSG_ENABLE_ZENOH.

OpenSSL is optional in the default CPM build: MSG_NATS_TLS=OFF keeps ordinary nats:// messaging available without TLS. Set MSG_NATS_TLS=ON to require TLS and OpenSSL 1.1.1+, or AUTO to enable TLS only when OpenSSL is found. System dependency builds inherit their installed NATS library’s TLS capabilities. Release link-time optimization is scoped to msg and its own tests and benchmarks; Windows builds respect CMake’s MSVC runtime selection.

Shared memory uses POSIX mappings and named semaphores on Linux and macOS, and named memory mappings and semaphores on Windows. Linux backend tests and Windows SHM URI integration tests have passed locally, including none, lz4, and zstd on Windows. Full Windows backend and macOS validation remain pending; CI runs on Linux. Process tests cover concurrent publishers, 64 readers, backpressure, ordering, and recovery after participant crashes.