Let sinks finish themselves and give the registry defaults and stopAll

CLAUDE.md

@@ -109,9 +109,11 @@ DATA_TAMER_UPDATE_GOLDEN=1 \
   `MCAPRingSink::requestDump()`. Tests assert it with `AllocCounter::Scope`
   (`tests/alloc_counter.hpp`): add one when you touch these paths. `takeSnapshot()` may
   block and grow a slot; keep the two variants distinct.
-- `onSchema()`/`onSnapshot()` are serialized by the SinkWorker, and control operations
-  (registration, sinks, `prepare()`) wait for them. Never call a control operation from a
-  sink callback, a serializer or inside `scopedWrite()`.
+- `onSchema()`/`onSnapshot()`/`onStop()` are serialized by the SinkWorker, and control
+  operations (registration, sinks, `prepare()`) wait for them. Never call a control
+  operation from a sink callback, a serializer or inside `scopedWrite()`. `stop()` runs
+  `onStop()` once per stop, after the last delivery; a sink that needs to finish
+  (close, flush, dump) does it there, not in a "call X after stop()" rule.
 - A change to the schema text, the payload encoding, the schema hash or the MCAP and ROS
   message layout is a format revision. Update `docs/wire_format.md`, regenerate the
   vectors, update `python/data_tamer_parser.py` and `data_tamer_parser.hpp`, and bump
@@ -120,8 +122,12 @@ DATA_TAMER_UPDATE_GOLDEN=1 \
 - LogChannel, ChannelsRegistry, SinkWorker, TypesRegistry, MCAPSink, MCAPRingSink and
   ROS2PublisherSink keep all their state behind a Pimpl (LogChannel's only other base is
   `enable_shared_from_this`). Add members to the Pimpl. `tests/abi_tests.cpp` pins the
-  sizes of the sinks, SinkWorker, LogChannel and SnapshotRef. Record any ABI break in the
-  CHANGELOG.
+  sizes of the sinks, SinkWorker, LogChannel, SnapshotRef, DataSink and ChannelDefaults.
+  Record any ABI break in the CHANGELOG.
+- The vtables of `DataSink` and `CustomSerializer` are frozen for 2.x: users derive from
+  them and the library calls through vtables compiled into their binaries. Add no virtual
+  function and no data member; new behaviour goes into non-virtual functions or a
+  separate interface.
 - Only `SnapshotPool::adopt()` builds a SnapshotRef from a pool slot; the constructor is
   private.
 - Include what you use. Headers that only name data_tamer types include `fwd.hpp`, which

README.md

@@ -35,8 +35,9 @@ a sink may save the information immediately in a file (currently, we support [MC
 or publish it using an inter-process communication, for instance, a ROS2 publisher.
 
 You can easily create your own, specialized sinks: implement `DataTamer::DataSink`
-(two callbacks, `onSchema` and `onSnapshot`) and wrap it with `SinkWorker::create<MySink>()`,
-which owns the delivery thread and the queues of the channels attached to it. See `data_tamer/sinks/dummy_sink.hpp` for a small one.
+(two callbacks, `onSchema` and `onSnapshot`, plus an optional `onStop` to finish at shutdown)
+and wrap it with `SinkWorker::create<MySink>()`, which owns the delivery thread and the queues
+of the channels attached to it. See `data_tamer/sinks/dummy_sink.hpp` for a small one.
 
 Headers that only name DataTamer types (a `LogChannel&` parameter, a `std::shared_ptr<SinkWorker>`
 member, ...) can include `data_tamer/fwd.hpp` instead of the full headers. Like `<iosfwd>`, it
@@ -189,6 +190,28 @@ Non-scalar `set()`/`get()` lock automatically and can be called inside
 them. Non-scalar pointer proxies also hold the write mutex; release them before
 starting a transaction or taking a snapshot.
 
+## Registry defaults and shutdown
+
+`ChannelsRegistry` creates channels by name, applies default settings to each new channel
+and attaches the default sinks to every channel, including the ones that exist already.
+`SinkWorker::stop()` delivers what is queued and then lets the sink finish itself
+(`DataSink::onStop()`): `MCAPSink` closes its file, `MCAPRingSink` writes a pending dump,
+`ROS2PublisherSink` publishes its partial batch. `stopAll()` does that for every sink of the
+registry, so shutdown needs no ordering rules.
+
+```cpp
+using namespace std::chrono_literals;
+auto& registry = DataTamer::ChannelsRegistry::Global();
+DataTamer::ChannelDefaults defaults;
+defaults.pool_stall_tolerance = 200ms;  // pool sized to absorb a 200 ms sink stall
+defaults.pool_snapshot_period = 1ms;    // at 1 kHz: 200 slots per channel
+registry.setChannelDefaults(defaults);  // channels created from now on
+registry.addDefaultSink(DataTamer::MCAPSink::create("run.mcap"));
+auto channel = registry.getChannel("controller");
+// ... register values, prepare(), take snapshots ...
+registry.stopAll();  // every sink stopped once; run.mcap is complete
+```
+
 ## How to register custom types
 
 Containers such as `std::vector` and `std::array` are supported out of the box.
@@ -318,7 +341,7 @@ options.schema_format = DataTamer::SchemaFormat::Yaml;  // optional: shorter sch
 
 auto sink = DataTamer::ROS2PublisherSink::create(node, "/robot", options);
 channel->addDataSink(sink);
-// a partial batch is published by the next snapshot, by flush() or when the sink is destroyed:
+// a partial batch is published by the next snapshot, by flush() or by sink->stop():
 sink->as<DataTamer::ROS2PublisherSink>().flush();
 ```
 
@@ -358,9 +381,8 @@ auto& recorder = worker->as<DataTamer::MCAPRingSink>();
 // From any thread, real-time ones included: one atomic compare-exchange.
 recorder.requestDump(std::chrono::seconds(2));  // also keep 2 s after the event
 
-// At shutdown, write a dump requested just before it.
+// At shutdown, stop() delivers what is queued and writes a dump requested just before it.
 worker->stop();
-recorder.flushPendingDump();
 ```
 
 - The trigger is the first snapshot delivered after the request, and all times are snapshot

data_tamer_cpp/CHANGELOG.rst

@@ -4,6 +4,32 @@ Changelog for package data_tamer
 
 Unreleased
 ----------
+* **Sinks finish themselves**: new ``virtual void DataSink::onStop()``,
+  empty by default. ``SinkWorker::stop()`` calls it on the stopping thread
+  after the last delivery, serialized with the other callbacks, once per stop
+  (a second ``stop()`` without ``start()`` does not call it again; the
+  destructor of a running worker does). A throw is counted in ``errors()``.
+  ``MCAPSink`` closes its file there (**behaviour change**: ``stop()`` now ends
+  the recording; call ``restartRecording()`` before ``start()`` to record
+  again), ``MCAPRingSink`` writes a pending dump (``flushPendingDump()``),
+  ``ROS2PublisherSink`` publishes its partial batch (``flush()``). The "call
+  ``stopRecording()`` / ``flushPendingDump()`` / ``flush()`` after ``stop()``"
+  rules are gone.
+* **ABI**: ``onStop()`` appends a slot to the ``DataSink`` vtable, so sinks
+  compiled against an earlier build must be rebuilt. From 2.0 on, the
+  ``DataSink`` and ``CustomSerializer`` vtables are frozen for 2.x (noted in
+  their headers); ``tests/abi_tests.cpp`` pins ``sizeof(DataSink)`` and
+  ``sizeof(ChannelDefaults)``.
+* ``ChannelsRegistry::stopAll()`` stops, once each, every default sink and
+  every sink attached to a channel the registry created.
+* ``ChannelsRegistry::addDefaultSink()`` also attaches the sink to the
+  channels that exist already (a prepared one announces its schema at once),
+  so it no longer has to come before ``getChannel()``. If a channel refuses the
+  sink, the call is undone and the exception propagates.
+* ``ChannelsRegistry::setChannelDefaults(ChannelDefaults)``: pool capacity (as
+  a count or in time) and payload capacity applied by ``getChannel()`` to the
+  channels it creates afterwards, before the default sinks are attached.
+  Existing channels are not changed. ``clear()`` resets them.
 * **Breaking, sinks**: the shared sink queue is gone, and with it
   ``SinkWorker::kDefaultQueueCapacity`` and the ``queue_capacity`` constructor
   argument: the constructor is ``SinkWorker(std::unique_ptr<DataSink>,
@@ -57,8 +83,8 @@ Unreleased
   completes at the first snapshot stamped after ``T + post_trigger``.
   A writer thread of the sink writes the file from a second buffer, so the
   sink worker never waits for disk; a finished dump that finds the writer busy
-  is handed over at the next snapshot. ``flushPendingDump()`` writes a request
-  made just before ``SinkWorker::stop()``. Dump ``N`` is written to
+  is handed over at the next snapshot. ``SinkWorker::stop()`` writes a request
+  made just before it (``flushPendingDump()``). Dump ``N`` is written to
   ``details::NumberedPath(filepath, N)``, skipping names that exist already.
   The dump callback reports write errors (disk full) and ``truncated`` when
   the ring was too small for the interval. Example ``T04_flight_recorder``.

data_tamer_cpp/examples/T01_basic_example.cpp

@@ -7,8 +7,8 @@ int main()
 {
   using namespace DataTamer;
 
-  // start defining one or more Sinks that must be added by default.
-  // Do addDefaultSink BEFORE creating a channel.
+  // Sinks added by default are attached to every channel of the registry,
+  // the existing ones and the ones created later.
   auto dummy_sink = DummySink::create();
   ChannelsRegistry::Global().addDefaultSink(dummy_sink);
 

data_tamer_cpp/examples/T04_flight_recorder.cpp

@@ -48,8 +48,8 @@ int main()
     std::this_thread::sleep_for(1ms);
   }
 
-  // Write a dump requested just before shutdown, if any, and wait for the writer.
+  // Deliver what is queued, then write a dump requested just before shutdown, if
+  // any, and wait for the writer.
   worker->stop();
-  recorder.flushPendingDump();
   return 0;
 }

data_tamer_cpp/examples/mcap_1m_per_sec.cpp

@@ -71,8 +71,7 @@ void WritingThread(const std::string& channel_name)
 
 int main()
 {
-  // Start defining one or more Sinks that must be added by default.
-  // Do this BEFORE creating a channel.
+  // Sinks added by default reach every channel of the registry.
   auto mcap_sink = MCAPSink::create("test_1M.mcap");
   ChannelsRegistry::Global().addDefaultSink(mcap_sink);
 

data_tamer_cpp/include/data_tamer/channel.hpp

@@ -10,6 +10,7 @@
 #include <cstdint>
 #include <memory>
 #include <mutex>
+#include <vector>
 
 namespace DataTamer
 {
@@ -366,6 +367,9 @@ class LogChannel : public std::enable_shared_from_this<LogChannel>
 private:
   template <typename T>
   friend class LoggedValue;
+  friend class ChannelsRegistry;
+  /// The attached sinks, for ChannelsRegistry (stopAll, addDefaultSink).
+  [[nodiscard]] std::vector<std::shared_ptr<SinkWorker>> dataSinks() const;
   /// State shared with this channel's LoggedValues (enable flags, write mutex).
   [[nodiscard]] std::shared_ptr<ChannelSharedState> sharedState() const;
 

data_tamer_cpp/include/data_tamer/custom_types.hpp

@@ -9,6 +9,15 @@
 namespace DataTamer
 {
 
+/**
+ * @brief Serializes a type the library does not know (registerCustomValue()).
+ *
+ * ABI: the virtual functions below are frozen for 2.x. The library calls them
+ * through vtables compiled into user binaries, so adding, removing or
+ * reordering one breaks every serializer built against an earlier 2.x release.
+ * New behaviour arrives as a separate interface; CustomSerializer keeps no
+ * data members.
+ */
 class CustomSerializer
 {
 public:

data_tamer_cpp/include/data_tamer/data_sink.hpp

@@ -74,7 +74,7 @@ class SnapshotRef
 };
 
 /**
- * @brief Interface implemented by a sink. The two callbacks are invoked only by
+ * @brief Interface implemented by a sink. The callbacks are invoked only by
  * the SinkWorker that owns the sink and are serialized with each other, so a
  * sink needs no lock of its own for the state they touch:
  *
@@ -83,6 +83,11 @@ class SnapshotRef
  * - onSnapshot() runs on the worker thread (or in drain()), in the order each
  *   channel took its snapshots. Throw to report a failure: the worker counts
  *   it and keeps the message (see SinkWorker).
+ * - onStop() runs on the thread calling SinkWorker::stop() (or destroying the
+ *   worker), after the last snapshot has been delivered: the place to close a
+ *   file, publish a partial batch or write a pending dump. It runs once per
+ *   stop: a second stop() without start() in between does not call it again.
+ *   A throw is counted like one from onSnapshot(). The default does nothing.
  *
  * Keep onSnapshot() short. Every queued or retained SnapshotRef holds a slot of
  * the channel's snapshot pool, which all sinks of that channel share
@@ -99,6 +104,12 @@ class SnapshotRef
  * A callback may use the channel's const queries (getSchema(), stats(), ...)
  * but must never call anything that changes it (registration, sinks,
  * prepare()): those wait for callbacks to finish and would deadlock.
+ *
+ * ABI: the virtual functions below are frozen for 2.x. The library calls them
+ * through vtables compiled into user binaries, so adding, removing or
+ * reordering one breaks every sink built against an earlier 2.x release.
+ * New behaviour arrives as non-virtual functions of SinkWorker or as a
+ * separate interface; DataSink keeps no data members.
  */
 class DataSink
 {
@@ -109,6 +120,7 @@ class DataSink
   friend class SinkWorker;
   virtual void onSchema(const Schema& schema) = 0;
   virtual void onSnapshot(const SnapshotRef& snapshot) = 0;
+  virtual void onStop() {}
 };
 
 /**
@@ -162,10 +174,15 @@ class SinkWorker
   }
 
   /// Stop accepting snapshots, wait for the callback in progress, join the
-  /// worker thread and deliver everything still queued. Idempotent. Never call
-  /// it from a callback.
+  /// worker thread, deliver everything still queued, then call the sink's
+  /// onStop() on this thread (MCAPSink closes its file, MCAPRingSink writes a
+  /// pending dump, ROS2PublisherSink publishes its partial batch). Idempotent:
+  /// calling it again before start() does not call onStop() a second time.
+  /// The destructor calls it. Never call it from a callback.
   void stop();
-  /// Resume after stop(): reopen admission and restart the worker thread.
+  /// Resume after stop(): reopen admission and restart the worker thread. The
+  /// sink is not told; a sink that finished itself in onStop() needs its own
+  /// restart (e.g. MCAPSink::restartRecording()) before start().
   void start();
   /// Deliver every snapshot queued so far on the calling thread. Returns after
   /// a callback in progress on the worker has finished, so everything taken
@@ -182,7 +199,8 @@ class SinkWorker
     return dynamic_cast<T&>(sink());
   }
 
-  /// Number of onSnapshot() calls that threw, and the message of the last one.
+  /// Number of onSnapshot() and onStop() calls that threw, and the message of
+  /// the last one.
   [[nodiscard]] uint64_t errors() const;
   [[nodiscard]] std::string lastError() const;
 

data_tamer_cpp/include/data_tamer/data_tamer.hpp

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <chrono>
+#include <cstddef>
 #include <memory>
 #include <string>
 
@@ -11,6 +13,26 @@
 namespace DataTamer
 {
 
+/**
+ * @brief Settings ChannelsRegistry::getChannel() applies to every channel it
+ * creates, before attaching the default sinks. Zero leaves a setting at the
+ * library default. Set the pool either as a count or in time, not both.
+ *
+ * ABI: passed by value across the library boundary; a change of its layout
+ * requires a SONAME bump.
+ */
+struct ChannelDefaults
+{
+  /// LogChannel::setPoolCapacity(count): pool slots per channel.
+  size_t pool_capacity = 0;
+  /// LogChannel::setPoolCapacity(stall_tolerance, snapshot_period): the pool
+  /// sized in time. Set both or neither.
+  std::chrono::nanoseconds pool_stall_tolerance{ 0 };
+  std::chrono::nanoseconds pool_snapshot_period{ 0 };
+  /// LogChannel::setPayloadCapacity(bytes): minimum payload bytes per slot.
+  size_t payload_capacity = 0;
+};
+
 class ChannelsRegistry
 {
 public:
@@ -22,14 +44,40 @@ class ChannelsRegistry
   // global instance (similar to singleton)
   static ChannelsRegistry& Global();
 
-  /// Once added here, this sink will be automatically connected to
-  /// any new channel created with getChannel()
+  /// Attach this sink to every channel of the registry: the ones that exist
+  /// (a prepared channel announces its schema to the sink at once, see
+  /// LogChannel::addDataSink) and the ones getChannel() creates later.
+  /// A sink a channel holds already is not attached twice. If a channel
+  /// refuses the sink (eight sinks already, onSchema() throws), the sink is
+  /// detached again from the channels this call attached it to, it does not
+  /// become a default sink, and the exception propagates.
   void addDefaultSink(std::shared_ptr<SinkWorker> sink);
 
-  /// Create a new channel or get a previously create one.
+  /// Settings for the channels getChannel() creates from now on. Channels that
+  /// exist already are not changed: their owner may have configured them, and
+  /// pool and payload sizes freeze at prepare().
+  /// Throws std::invalid_argument when both pool forms are set, or only one of
+  /// the two durations, and whatever the LogChannel setters throw for the
+  /// values (checked here, not at getChannel()); the previous defaults are
+  /// then kept.
+  void setChannelDefaults(const ChannelDefaults& defaults);
+
+  /// Create a new channel or get a previously created one. A new channel gets
+  /// the channel defaults, then the default sinks.
   [[nodiscard]] std::shared_ptr<LogChannel> getChannel(std::string const& channel_name);
 
-  /// remove all channels and stored sinks
+  /// SinkWorker::stop() on every sink the registry reaches, each once even
+  /// when several channels share it: the default sinks, and every sink
+  /// attached at the time of the call to a channel created by getChannel(),
+  /// whether by addDefaultSink() or by LogChannel::addDataSink(). Sinks of
+  /// channels made with LogChannel::create() are reached only if they are
+  /// default sinks. The stops run on the calling thread, in no particular
+  /// order; each delivers what is queued, then finishes the sink
+  /// (DataSink::onStop()). The sinks stay attached: restart one with
+  /// SinkWorker::start(). Never call it from a sink callback.
+  void stopAll();
+
+  /// Remove all channels and stored sinks, and reset the channel defaults.
   void clear();
 
 private:

data_tamer_cpp/include/data_tamer/sinks/mcap_ring_sink.hpp

@@ -110,8 +110,10 @@ struct MCAPRingStats
  * memcpy of up to capacity_bytes.
  *
  * Shutdown. A request made just before shutdown has no snapshot left to
- * trigger it: call flushPendingDump() after SinkWorker::stop(). The destructor
- * finishes the file being written but does not start a pending dump.
+ * trigger it: SinkWorker::stop() (and the worker's destructor) writes it with
+ * flushPendingDump() once the last snapshot is delivered, and waits for the
+ * writer. The sink's own destructor finishes the file being written but does
+ * not start a pending dump.
  */
 class MCAPRingSink : public DataSink
 {
@@ -146,9 +148,10 @@ class MCAPRingSink : public DataSink
    * @brief Write the active request, if any, with what the ring holds now,
    * and wait until it is on disk. A request no snapshot has triggered yet uses
    * the newest snapshot timestamp seen as trigger; a dump still waiting for
-   * its post-trigger interval is cut at the newest timestamp seen. Call it after
-   * SinkWorker::stop() (or between drain() calls with manual delivery).
-   * Never call it from the dump callback (std::logic_error).
+   * its post-trigger interval is cut at the newest timestamp seen.
+   * SinkWorker::stop() calls it after the last delivery; call it yourself only
+   * while no snapshot is delivered, e.g. between drain() calls with manual
+   * delivery. Never call it from the dump callback (std::logic_error).
    * Returns true if it wrote a dump (see MCAPRingDump::ok for the outcome).
    * In every case it returns after the writer has finished all dumps handed
    * to it so far.
@@ -169,6 +172,8 @@ class MCAPRingSink : public DataSink
 protected:
   void onSchema(const Schema& schema) override;
   void onSnapshot(const SnapshotRef& snapshot) override;
+  /// Writes a pending dump, as flushPendingDump().
+  void onStop() override;
 
 private:
   struct Pimpl;