safe-shm

Lock-free shared memory primitives for real-time C++ and Python.

Header-only C++23 library providing seqlock, cyclic ring buffer, time-series temporal queries, and lifecycle management — all over POSIX shared memory with zero-copy cross-process communication. Uses Linux futex for blocking waits, std::atomic_ref for lock-free access, and per-slot seqlocks to eliminate contention.

Works on x86_64 and ARM64 Linux.

Dependencies

• shm — POSIX shared memory wrapper
• exception-rt — runtime exception utilities
• fmt — formatting
• nanobind (optional) — Python bindings
• doctest (test only) — fetched automatically
• nanobench (bench only) — fetched automatically

Components

Core Primitives

Legacy / Specialized

Architecture

Seqlock (single value, lock-free)

Writer                                  Reader
┌────────────────────┐                 ┌────────────────────┐
│ seq++ (odd)        │                 │ read seq (acquire)  │
│ memcpy data        │  /dev/shm/name  │ memcpy data         │
│ seq++ (even)       ├────────────────►│ read seq (acquire)  │
│ futex_wake         │                 │ if changed → retry  │
└────────────────────┘                 └────────────────────┘

CyclicBuffer (ring buffer, per-slot seqlock)

Writer                                     Reader
┌──────────────────────┐                  ┌──────────────────────────┐
│ idx = writes & (N-1) │                  │ get_latest()             │
│ slot[idx].seq++ (odd)│  /dev/shm/name   │ get(reverse_index)       │
│ memcpy slot[idx].data├─────────────────►│ try_get() → optional<T>  │
│ slot[idx].seq++ (even│                  │ wait_for_write() [futex] │
│ total_writes++       │                  │                          │
│ futex_wake           │                  │ TimeSeries layer:        │
└──────────────────────┘                  │  find_closest(key)       │
                                          │  find_interpolation_pair │
  SHM layout:                             │  get_latest_if_fresh     │
  [total_writes] [slot₀] [slot₁] .. [slotₙ₋₁]                      │
  Each slot: [seq_counter | T data]       └──────────────────────────┘

Lifecycle Management

OwnedShm<T>
┌──────────────────────────────────┐
│ ShmHeader (32 bytes)             │
│   magic: 0x53484D48             │
│   version, heartbeat_ns         │
│   writer_pid                     │
├──────────────────────────────────┤
│ T data (user payload)            │
└──────────────────────────────────┘
  RAII: unlinks /dev/shm on destruction
  is_writer_alive(): PID check + heartbeat

Choosing a Transport

Quick Start

Seqlock (cross-process, lock-free)

#include "safe-shm/seqlock.hpp"

// Writer process
safe_shm::SeqlockWriter<SensorData> writer("sensor_shm");
writer.store(SensorData{20.5, 1013.25, 45.0, now_ms()});

// Reader process
safe_shm::SeqlockReader<SensorData> reader("sensor_shm");
auto data = reader.load();                              // non-blocking
auto data2 = reader.load_blocking(last_seq, timeout);   // blocks until new data

CyclicBuffer + TimeSeries (ring buffer with temporal queries)

#include "safe-shm/cyclic_buffer.hpp"
#include "safe-shm/stamped.hpp"
#include "safe-shm/time_series.hpp"

using StampedIMU = safe_shm::Stamped<IMUReading>;

// Writer: insert stamped readings at 200 Hz
safe_shm::CyclicBufferWriter<StampedIMU, 64> writer("imu_shm");
writer.insert(safe_shm::stamp(imu_reading, seq++));

// Reader: temporal queries
safe_shm::TimeSeries<StampedIMU, 64> ts("imu_shm");

auto latest = ts.get_latest();
auto closest = ts.find_closest(target_ns);           // binary search O(log N)
auto closest = ts.find_closest(target_ns, 100'000);  // with max_distance guard
auto interp = ts.find_interpolation_pair(target_ns);  // {before, after, alpha}
auto fresh = ts.get_latest_if_fresh(min_timestamp);    // staleness check

Lifecycle Management

#include "safe-shm/shm_lifecycle.hpp"

// RAII-managed SHM segment with header
safe_shm::OwnedShm<Config> owned("my_config");
owned.data() = {1.5, 0.0, 3};
owned.update_heartbeat();

// Remote liveness check
if (safe_shm::is_writer_alive(header, /*max_stale_ns=*/1'000'000'000))
    fmt::print("Writer is alive\n");

// Cleanup utilities
for (auto const& seg : safe_shm::shm_list("stale_"))
    safe_shm::shm_remove(seg);

Python Bindings

import safe_shm_py as shm

# Seqlock
writer = shm.SeqlockWriterF64("my_shm")
writer.store(3.14159)

reader = shm.SeqlockReaderF64("my_shm")
value = reader.load()  # 3.14159

# CyclicBuffer + TimeSeries
cb_writer = shm.CyclicBufferWriterStampedF64("ring_shm")
s = shm.StampedF64()
s.timestamp_ns = shm.monotonic_now_ns()
s.sequence = 0
s.data = 42.0
cb_writer.insert(s)

ts = shm.TimeSeriesStampedF64("ring_shm")
closest = ts.find_closest(target_ns)
interp = ts.find_interpolation_pair(target_ns)

Temporal Query Safety

Two layers of protection against unit/clock-type mismatches:

Layer 1: Runtime — max_distance parameter

// BUG: querying with milliseconds (5000) instead of nanoseconds
auto bad = ts.find_closest(5000);               // silently returns oldest element
auto bad = ts.find_closest(5000, 100'000'000);  // returns nullopt — mismatch caught!

// CORRECT: consistent nanosecond units
auto good = ts.find_closest(5'000'000'000, 100'000'000);  // works, within 100ms

Layer 2: Compile-time — SanitizedKey<Tag> (opt-in)

SanitizedKey<Tag> is a strong type that can only be created through an external validator function. The compiler prevents passing raw uint64_t where SanitizedKey is expected.

#include "safe-shm/sanitized_key.hpp"

// Step 1: Sanitize — the ONLY way to create a SanitizedKey
auto key = safe_shm::sanitize<safe_shm::MonotonicNsTag>(
    raw_ns, safe_shm::MonotonicNsValidator{});

if (!key) { /* rejected: wrong units or clock type */ }

// Step 2: Pass to TimeSeries — type-safe overload
auto result = ts.find_closest(*key);
auto interp = ts.find_interpolation_pair(*key);
auto fresh  = ts.get_latest_if_fresh(*key);

Built-in validators:

Different Tag types make different sanitization policies type-incompatible at compile time — SanitizedKey<MonotonicNsTag> cannot be passed where SanitizedKey<RangeCheckedTag> is expected.

Custom validators are just callables:

auto key = safe_shm::sanitize<MyTag>(raw_ns, [latest_ts](uint64_t v) {
    return (v > latest_ts ? v - latest_ts : latest_ts - v) <= 2'000'000'000ULL;
});

The raw uint64_t API is unchanged — SanitizedKey is purely opt-in.

Build

mkdir build && cd build
cmake .. -DCMAKE_BUILD_TYPE=Release
make -j$(nproc)
ctest

Sanitizers

cmake .. -DSANITIZER=asan    # Address + UB (default)
cmake .. -DSANITIZER=tsan    # Thread sanitizer
cmake .. -DSANITIZER=none    # No sanitizer

Fuzz Testing

# Requires clang
CC=clang CXX=clang++ cmake .. -DSANITIZER=fuzzer
make fuzz_cyclic_buffer fuzz_time_series
./fuzz_cyclic_buffer corpus_cyclic/ -max_total_time=60
./fuzz_time_series corpus_ts/ -max_total_time=60

Docker (CI)

docker build -t safe-shm-test -f Dockerfile.test .
docker run --rm -v $(pwd):/src safe-shm-test bash -c \
  "cd /tmp && cmake /src -DSANITIZER=none && make -j\$(nproc) && ctest"

Examples

All examples are buildable targets — see examples/ directory.

# C++ examples (two terminals)
./seqlock_example writer    # terminal 1
./seqlock_example reader    # terminal 2

# Python examples
PYTHONPATH=build python3 examples/python_seqlock.py
PYTHONPATH=build python3 examples/python_cyclic_buffer.py

Tests

Benchmarks

Run with: ./benchmark && ./benchmark_competitors (built with -O3 -DNDEBUG, nanobench)

Throughput (store + load cycle)

TimeSeries Query Latency

Key Takeaways

• CyclicBuffer at FHD: 39 GB/s — matches raw memcpy with full ring buffer history
• Seqlock vs CyclicBuffer: equivalent at small payloads, CyclicBuffer slightly faster at 4 KB
• vs kernel IPC: 2.7x faster than pipe at 64 B, 60x faster than POSIX mq at 4 KB
• TimeSeries lookups: 45 ns per binary search across 64 elements — suitable for real-time control loops
• Zero-copy: no intermediate buffers, no kernel transitions, no thread wakeups on read path

Design Decisions

Per-slot seqlock — Each CyclicBuffer slot has its own 32-bit sequence counter. Readers on slot i never contend with writers on slot j. The alternative (single global seqlock) forces retry on any concurrent write.

Power-of-two capacity — N & (N-1) == 0 enables bitmask modulo (idx & (N-1)) instead of expensive integer division. Enforced at compile time via static_assert.

Stamped<T> composability — Stamped<T> is itself a FlatType, so Seqlock<Stamped<SensorData>>, CyclicBuffer<Stamped<IMU>, 64>, and OwnedShm<Stamped<Config>> all work without adapter code.

std::optional return types — try_get(), find_closest(), find_interpolation_pair() return std::optional<T> instead of throwing. Compatible with std::expected patterns and real-time constraints (no heap allocation).

Futex for blocking — wait_for_write() and load_blocking() use raw FUTEX_WAIT/FUTEX_WAKE (not _PRIVATE) for cross-process signaling. Uncontended path is a single atomic compare — no syscall.

License

MIT