Repository navigation
Let sinks finish themselves and give the registry defaults and stopAll - #17
Merged
Merged
Conversation
Problem: shutting down needed an ordering rule per sink. Callers had to run MCAPSink::stopRecording(), MCAPRingSink::flushPendingDump() or ROS2PublisherSink::flush() after SinkWorker::stop(), or lose the end of a file, a pending dump or a partial batch. ChannelsRegistry reached only the channels created after addDefaultSink(), offered no way to stop its sinks, and could not size the channels it created, so consumers wrapped it. Change: - New virtual DataSink::onStop(), empty by default. SinkWorker::stop() calls it on the stopping thread after the final drain, under the same locks as onSnapshot() and onSchema(), once per stop: a second stop() without start() skips it, the destructor of a running worker runs it. A throw is counted in errors(). MCAPSink closes its file, MCAPRingSink runs flushPendingDump(), ROS2PublisherSink runs flush(). - ChannelsRegistry::stopAll() stops, once each, every default sink and every sink attached to a channel the registry created. - addDefaultSink() also attaches the sink to existing channels (prepared ones announce their schema through addDataSink()), and undoes the call if a channel refuses the sink. - setChannelDefaults(ChannelDefaults): pool capacity as a count or in time, and payload capacity, validated when set and applied by getChannel() to new channels before the default sinks. Existing channels are left alone. clear() resets the defaults. - DataSink and CustomSerializer headers state that their vtables are frozen for 2.x. abi_tests pins sizeof(DataSink) and sizeof(ChannelDefaults). Compatibility: the onStop() slot changes the DataSink vtable, so sinks built against an earlier build must be rebuilt (2.0 is untagged). MCAPSink stops recording at SinkWorker::stop(); call restartRecording() before start() to record again. The new registry functions are additive. RT impact: none. tryTakeSnapshot() and SinkWorker::tryPush() are unchanged; onStop() and the registry functions run on control threads. Tested in the jazzy container: colcon (msgs + cpp, tests and examples), presets debug, asan and tsan, Python tests, and a TSan loop of the lifecycle, registry, ring and queue tests (x100). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WASypwgBSXDzJkPN4HBtFD
facontidavide
pushed a commit
that referenced
this pull request
Oct 7, 2026
…emselves Every channel published into one moodycamel queue per SinkWorker; explicit producers keep the blocks they take, so a channel could be rejected while its pool still had free slots, and users had to size queue_capacity by hand. Each (channel, sink) attachment now has a single-producer single-consumer ring of exactly the pool's capacity: the pool is the only bound. The worker polls 20 us, then sleeps on an eventcount with a std::binary_semaphore; a push costs one increment and one load and wakes the worker only after it slept. - Removed: queue_capacity, kDefaultQueueCapacity, the moodycamel names in data_sink.hpp and the vendored 3rdparty/ConcurrentQueue, LogChannel::getActiveFlags(), the public SnapshotRef(pool, slot). - LogChannel::setPoolCapacity(stall, period); exact payload reserve for fixed-size schemas; addDataSink() returns whether it attached. - DataSink::onStop() and onStart(): sinks close files, write pending dumps and flush batches when the worker stops, and MCAPSink reopens on start. The DataSink and CustomSerializer vtables are frozen for 2.x. - ChannelsRegistry: stopAll(), setChannelDefaults(ChannelDefaults), addDefaultSink() also reaches existing channels (undone on refusal). - SinkWorker and LogChannel keep their sink and type registry in the Pimpl; abi_tests pins the layouts. - README, LLM manual, CLAUDE.md, header comments and CHANGELOG updated. Reviewed twice (memory ordering, wake-ups, attach/detach, lifecycle) and simplified; see PR #16 and #17. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WASypwgBSXDzJkPN4HBtFD
facontidavide
pushed a commit
that referenced
this pull request
Oct 7, 2026
…emselves Every channel published into one moodycamel queue per SinkWorker; explicit producers keep the blocks they take, so a channel could be rejected while its pool still had free slots, and users had to size queue_capacity by hand. Each (channel, sink) attachment now has a single-producer single-consumer ring of exactly the pool's capacity: the pool is the only bound. The worker polls 20 us, then sleeps on an eventcount with a std::binary_semaphore; a push costs one increment and one load and wakes the worker only after it slept. - Removed: queue_capacity, kDefaultQueueCapacity, the moodycamel names in data_sink.hpp and the vendored 3rdparty/ConcurrentQueue, LogChannel::getActiveFlags(), the public SnapshotRef(pool, slot). - LogChannel::setPoolCapacity(stall, period); exact payload reserve for fixed-size schemas; addDataSink() returns whether it attached. - DataSink::onStop() and onStart(): sinks close files, write pending dumps and flush batches when the worker stops, and MCAPSink reopens on start. The DataSink and CustomSerializer vtables are frozen for 2.x. - ChannelsRegistry: stopAll(), setChannelDefaults(ChannelDefaults), addDefaultSink() also reaches existing channels (undone on refusal). - SinkWorker and LogChannel keep their sink and type registry in the Pimpl; abi_tests pins the layouts. - README, LLM manual, CLAUDE.md, header comments and CHANGELOG updated. Reviewed twice (memory ordering, wake-ups, attach/detach, lifecycle) and simplified; see PR #16 and #17. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WASypwgBSXDzJkPN4HBtFD
facontidavide
added a commit
that referenced
this pull request
Oct 7, 2026
…emselves (#16) Every channel published into one moodycamel queue per SinkWorker; explicit producers keep the blocks they take, so a channel could be rejected while its pool still had free slots, and users had to size queue_capacity by hand. Each (channel, sink) attachment now has a single-producer single-consumer ring of exactly the pool's capacity: the pool is the only bound. The worker polls 20 us, then sleeps on an eventcount with a std::binary_semaphore; a push costs one increment and one load and wakes the worker only after it slept. - Removed: queue_capacity, kDefaultQueueCapacity, the moodycamel names in data_sink.hpp and the vendored 3rdparty/ConcurrentQueue, LogChannel::getActiveFlags(), the public SnapshotRef(pool, slot). - LogChannel::setPoolCapacity(stall, period); exact payload reserve for fixed-size schemas; addDataSink() returns whether it attached. - DataSink::onStop() and onStart(): sinks close files, write pending dumps and flush batches when the worker stops, and MCAPSink reopens on start. The DataSink and CustomSerializer vtables are frozen for 2.x. - ChannelsRegistry: stopAll(), setChannelDefaults(ChannelDefaults), addDefaultSink() also reaches existing channels (undone on refusal). - SinkWorker and LogChannel keep their sink and type registry in the Pimpl; abi_tests pins the layouts. - README, LLM manual, CLAUDE.md, header comments and CHANGELOG updated. Reviewed twice (memory ordering, wake-ups, attach/detach, lifecycle) and simplified; see PR #16 and #17. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WASypwgBSXDzJkPN4HBtFD
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #16. Merge #16 first.
Sinks now finish their own work when they stop. The registry also gets defaults for new channels and a way to stop everything.
What changes
DataSink::onStop(). An empty virtual thatSinkWorker::stop()calls on the stopping thread after the final drain, serialized with the other sink callbacks.stop()withoutstart()skips it, and destroying a running worker calls it.errors()and never escapesstop().MCAPSinkcloses its file.MCAPRingSinkwrites a pending dump (flushPendingDump()).ROS2PublisherSinkflushes its batch.stop().ChannelsRegistry::stopAll(). Stops each worker once: the default sinks and every sink attached to a channel the registry created. It runs outside the registry lock.addDefaultSink()reaches existing channels. Prepared channels get the schema through the existingaddDataSink()path. If one channel refuses the sink (eight sinks already, oronSchema()throws), the whole call is undone and the exception propagates.setChannelDefaults(ChannelDefaults). Sets pool capacity, as a count or in time (SinkWorker: one queue per channel, sized by the pool #16), and payload capacity.getChannel()applies them to new channels before attaching the default sinks.clear()resets them.DataSinkandCustomSerializerheaders now state that their vtables are frozen for 2.x.abi_testspins the sizes ofDataSinkandChannelDefaults.Behaviour change
MCAPSink::stop()now ends the recording. To record again, callrestartRecording()(a new file) beforestart(). Code that used stop and start as a pause within the same file has to change. Nothing in this repository does that.Real-time path
Unchanged:
tryTakeSnapshot()andtryPush()are the same as in #16.Testing
After rebasing onto #16 with its review fixes, the debug preset passes 210/210.
The MCAP finalize test fails if
MCAPSink::onStop()is emptied.Known limits of the undo in
addDefaultSink():addDataSink()of the same sink on the same channel could be undone too.🤖 Generated with Claude Code
https://claude.ai/code/session_01WASypwgBSXDzJkPN4HBtFD