Real-time friendly snapshot path: bounded allocation-free sink queue, tryTakeSnapshot(), startRecording()

[facontidavide]

Addresses the real-time items of #73 (blocking locks, per-tick heap allocations, sink setup on the first tick, unbounded queue). The std::array compile error and the schema hash gap from that issue are left for separate PRs.

What changes

DataSinkBase: bounded, allocation-free handoff (src/data_sink.cpp)

• The ConcurrentQueue<Snapshot> is replaced by a pool of queue_size pre-allocated Snapshot slots and two ConcurrentQueue<Snapshot*> (free list and ready list) that never allocate on try_enqueue/try_dequeue.
• pushSnapshot() takes a free slot, copies with vector::assign (no allocation once the slot's capacity covers the payload) and pushes the pointer. If no slot is free the snapshot is dropped and counted; memory never grows.
• New constructor DataSinkBase(size_t queue_size, size_t reserved_payload_bytes = 0); default constructor keeps kDefaultQueueSize = 64. New getters queueSize() and droppedSnapshotsCount().

LogChannel::tryTakeSnapshot() (src/channel.cpp)

• Same as takeSnapshot() but try_locks the channel mutex and returns false if another thread holds it. Dropping a sample is the failure mode a control loop wants.
• The sink list is now copy-on-write (shared_ptr<const vector> swapped with std::atomic_store). Neither snapshot variant takes sinks_mutex anymore, so a concurrent addDataSink() doing I/O in addChannel() cannot stall the producer. addDataSink() also dedupes, matching the previous unordered_set semantics.
• The push loop now runs while the channel mutex is still held, which protects _p->snapshot against two threads snapshotting the same channel (previously a race).

LogChannel::startRecording()

• Freezes the schema and calls addChannel() on all sinks. Optional: the first snapshot still does this lazily, but a real-time caller can now do it at setup time. Idempotent; sinks added afterwards get the schema immediately, as before.

README: new "Real-time usage" section.

Behaviour changes to be aware of

• A sink that cannot keep up now drops the newest snapshots after 64 queued (configurable) instead of growing the queue without bound. pushSnapshot() already documented false as "queue full", so the contract is unchanged, but the outcome is now reachable.
• The first pushSnapshot() from a new thread lets the queue register that thread as a producer, which may allocate once. Pre-allocation covers up to 4 producer threads per sink.

Tests

tests/realtime_tests.cpp, four cases:

• startRecording() registers the schema before any snapshot, freezes registration, is idempotent, and late sinks still get the schema.
• tryTakeSnapshot() returns false while writeMutex() is held by the test, true afterwards.
• With a consumer blocked inside storeSnapshot() and queue_size = 4, 10 pushes yield exactly 4 accepted and droppedSnapshotsCount() == 6; after release, pushing works again.
• Steady-state tryTakeSnapshot() on a channel with a 100-double vector, a std::array<float,16> and two scalars performs zero heap allocations on the calling thread, measured by a replaced global operator new gated by a thread_local flag.

Full suite (28 tests) passes in Debug and Release (-Wall -Wextra, no new warnings). The RealTime.* tests pass 50 consecutive repeats. Not run here: the ROS 2 build (ros2_publisher_sink is untouched and only uses the DataSinkBase interface).

🤖 Generated with Claude Code