SinkWorker: one queue per channel, sized by the pool

[facontidavide]

Replaces the shared sink queue with one single-producer single-consumer ring per (channel, sink) attachment. Each ring has exactly the channel's pool capacity, so a snapshot can no longer be refused while its pool has a free slot. The pool becomes the only bound, and queue sizing disappears from the API.

Why

Every channel published into one moodycamel queue per SinkWorker. An explicit producer keeps the 32-entry blocks it takes, so with the default 1024 entries a channel could get rejected while its pool still had free slots. Users had to size queue_capacity from channel counts and block arithmetic.

What changes

• Per-attachment rings.
  • A ring is allocated with the pool in prepare(), or in addDataSink() on a prepared channel.
  • Detaching marks the ring under the channel's control mutex. The worker delivers what it holds, before any newer attachment of the same channel, then frees it. No data is lost, and no stale entries reach the next channel.
  • Order is FIFO per channel. There is no order across channels, which matches what the docs already said.
• Worker wake-up.
  • An idle worker polls for 20 µs, then sleeps on a lock-free eventcount with a std::binary_semaphore.
  • A push costs one increment and one load. It wakes the worker with a futex call only when the worker has slept, and a burst of pushes costs at most one wake.
  • std::atomic::wait is avoided because, in libstdc++, it can turn the existing per-snapshot notify_all() calls into futex wakes.
• Removed.
  • SinkWorker::kDefaultQueueCapacity and the queue_capacity argument.
  • The moodycamel names in data_sink.hpp.
  • The vendored 3rdparty/ConcurrentQueue, which nothing uses any more.
  • LogChannel::getActiveFlags().
  • The public SnapshotRef(pool, slot) constructor. It's now private and built by SnapshotPool::adopt().
• Pool in time. New setPoolCapacity(stall_tolerance, snapshot_period) that computes ceil(stall / period) slots. It's validated and overflow-safe.
• Exact payload reserve. When every value is fixed-size (ValuePtr::isFixedSize()), each slot reserves exactly the payload. Otherwise it keeps max(2x, 256). setPayloadCapacity() stays a floor.
• ABI. SinkWorker keeps its sink in the Pimpl and LogChannel keeps its TypesRegistry in the Pimpl. abi_tests pins both sizes. 2.0.0 is not tagged yet.
• Docs. README, the LLM manual, CLAUDE.md, header comments and the CHANGELOG are updated.

Real-time path

No new allocation, lock or I/O.

Futex calls on the RT thread, measured with strace on a laptop over 2000 ticks:

Build	1 channel	4 channels
V2 before this PR	2001	2828
This PR	2035	2614

The poll costs up to 2 % of a core per worker at 1 kHz. This hasn't been measured on an embedded target yet.

Review

An independent review covered:

• memory ordering;
• lost wake-ups;
• attach and detach during logging;
• the never-rejected invariant;
• the RT path.

It found 2 important issues. Both are fixed:

• Futex wakes on bursts. The worker now polls before it sleeps.
• Order across concurrent remove and add. The detach now happens under the lock. The new two-thread test also caught a second ordering bug: detached rings drained newest first. That's fixed too.

Testing

Build	Result
colcon, ROS 2 Jazzy	209 gtests + 2 ctest entries, 0 failures
plain debug	196/196
asan	196/196
tsan	196/196
TSan stress: queue tests + ChannelCapacity.TenThousand*, x50	pass
debug stress: queue tests, x200	pass
ASan stress: whole suite, x20	pass
Python	OK

New tests:

• with 4 channels x 300 slots, no snapshot is rejected across 64 rounds or after detach and attach, and there are zero allocations;
• concurrent attach and detach keeps every snapshot and the per-channel order;
• remove and add on two threads keep the order;
• drain() frees detached rings;
• a wake-up stress test of 20000 ping-pongs around the poll window;
• the pool in time;
• the exact payload reserve, including custom types with vector members.

🤖 Generated with Claude Code

https://claude.ai/code/session_01WASypwgBSXDzJkPN4HBtFD