diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c5420f..e928ded 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 receive their slots as typed `std::optional`s (`std::nullopt` for a hole), with no `any_cast` in user code. `MergeFilter` / `registerMergeFilter` declare no slot types and keep their current, unchecked behaviour. +- **`MessageFilter::finish()`** — end-of-stream hook, called once after the last + message, so that a stage holding state can flush it (write a report, close a + file, publish a result to the `GraphContext`). It defaults to a no-op and + produces no message. `DslFilterGraph::finish()` finishes every stage in run + order; `FilterGraph`, `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter`, + `JoinFilter` and nested graphs forward it. Every stage is finished even if one + throws; the first exception is rethrown afterwards. There is no `tick()`: use + an in-band tick message (see README). - **`MergeStage` / `MergeSlotTypes`** (`MergeStage.hpp`, new header) — how a merge stage declares its slot types; `AnyMessageFilter::mergeInputTypes()` exposes them to the DSL. `MergeInputs` moved here from `MergeFilter.hpp` @@ -51,9 +59,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the `std::shared_ptr` for composites that forward it. ### Changed -- **Breaking:** `AnyMessageFilter` has a new pure virtual - `setContext(std::shared_ptr)`; custom implementations must - forward the context to the stages they wrap. +- **Breaking:** `AnyMessageFilter` has new pure virtuals + `setContext(std::shared_ptr)` and `finish()`; custom + implementations must forward both to the stages they wrap. They are pure + rather than no-ops on purpose: a composite that forgot to forward them would + otherwise fail silently. - README, EXAMPLE.md and `apps/textPipeline` now describe runtime graphs in the DSL first; the JSON format is documented as a supported alternative. - `dsl::toMermaid` now uses generated node ids (instead of edge names such as diff --git a/README.md b/README.md index 05c99cb..aac00ae 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,9 @@ drops the message, and everything downstream of that edge is skipped. - **`GraphContext`** — a graph-scoped, type-keyed, thread-safe blackboard, so stages can share values without knowing who produced them. See [Graph context](#graph-context). +- **`MessageFilter::finish()`** — the end-of-stream hook: a stage that + accumulates state flushes it when the graph is finished. See + [Ending a run](#ending-a-run). ### Runtime graphs in the text DSL @@ -439,6 +442,57 @@ Custom composite stages override it to forward the context to their inner stages; `MessageFilter::sharedContext()` returns the `std::shared_ptr` to pass on. +## Ending a run + +A stage acts when a message reaches it, so a stage that accumulates something +(statistics, a batch, an open file) has no natural point at which to flush it. +`MessageFilter::finish()` is that point: + +```cpp +class WriteReport : public MessageFilter +{ +public: + std::optional filter(Record&& record) override + { + mSeen.push_back(record); + return std::move(record); + } + + void finish() override // called once, after the last message + { + writeJson(mPath, mSeen); + } + +private: + std::vector mSeen; + std::filesystem::path mPath; +}; +``` + +```cpp +DslFilterGraph graph(text); +for (auto&& record : records) { graph.filter(std::move(record)); } +graph.finish(); // finishes every stage, in run order +``` + +- `finish()` defaults to a no-op, so existing stages are unaffected. +- Every graph type and composite stage (`FilterGraph`, `DslFilterGraph`, + `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter`, `JoinFilter`, nested + graphs) forwards it to its stages. A `DslFilterGraph` finishes its stages in + run order, so a stage is finished after the stages it reads from. +- **Every stage is finished even if one of them throws**; the first exception is + rethrown once the others have run. +- `finish()` produces no message, so nothing is routed downstream. A stage that + wants to hand a final result to the application publishes it through the + [graph context](#graph-context) (or its own side channel). +- Calling `finish()` is the owner's decision: a graph that is never finished + never flushes, and finishing twice finishes every stage twice. + +There is no `tick()`. Time is domain-specific (event time, wall clock, a sensor +clock), so a stage that must act while no messages arrive is better served by an +**in-band tick message**: make the graph's input type a variant with a `Tick` +alternative and feed ticks through the graph like any other message. + ## JSON format The JSON format predates the DSL and remains fully supported; both use the diff --git a/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp b/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp index b654a3b..74024d7 100644 --- a/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp +++ b/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp @@ -103,6 +103,16 @@ class AnyFilterChain : public AnyMessageFilter } } + void finish() override + { + detail::FinishScope finished; + for (auto& stage : mStages) + { + finished.run([&stage] { stage->finish(); }); + } + finished.rethrow(); + } + private: std::vector> mStages; // Shared by the stages of this chain until a graph hands down its own, so diff --git a/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp b/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp index 8556993..4be3cd4 100644 --- a/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp +++ b/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp @@ -32,6 +32,10 @@ class AnyMessageFilter // MessageFilter::setContext. virtual void setContext(std::shared_ptr context) = 0; + // Finishes the wrapped stage(s); see MessageFilter::finish. Pure, like + // setContext, so that a composite cannot silently fail to forward it. + virtual void finish() = 0; + // The slot types of a merge stage (see MergeStage); empty for an untyped // merge and for every stage that is not a merge. virtual MergeSlotTypes mergeInputTypes() const @@ -77,6 +81,11 @@ class AnyMessageFilterAdapter : public AnyMessageFilter mFilter->setContext(std::move(context)); } + void finish() override + { + mFilter->finish(); + } + MergeSlotTypes mergeInputTypes() const override { if (const auto* merge = dynamic_cast(mFilter.get())) diff --git a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp index 52a86e7..beda7e8 100644 --- a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp @@ -484,6 +484,18 @@ class GraphPlan } } + // Finishes every stage in run order, so that a stage sees its upstream + // stages finished before itself. + void finish() + { + filterGraph::detail::FinishScope finished; + for (auto& stage : mStages) + { + finished.run([&stage] { stage.filter->finish(); }); + } + finished.rethrow(); + } + private: static constexpr std::size_t kInputSlot = 0; @@ -636,6 +648,14 @@ class DslFilterGraph : public MessageFilter MessageFilter::setContext(std::move(context)); } + // Finishes every stage of the graph, in run order; nested graphs are + // stages, so they finish their own stages in turn. See + // MessageFilter::finish. + void finish() override + { + mPlan.finish(); + } + // The parsed graph, e.g. for dsl::toMermaid. const dsl::GraphProgram& program() const { diff --git a/libs/filterGraph/core/filterGraph/FanoutFilter.hpp b/libs/filterGraph/core/filterGraph/FanoutFilter.hpp index b34c4c1..e6e36e4 100644 --- a/libs/filterGraph/core/filterGraph/FanoutFilter.hpp +++ b/libs/filterGraph/core/filterGraph/FanoutFilter.hpp @@ -66,6 +66,16 @@ class FanoutFilter : public MessageFilter MessageFilter::setContext(std::move(context)); } + void finish() override + { + detail::FinishScope finished; + for (auto& receiver : mReceivers) + { + finished.run([&receiver] { receiver->finish(); }); + } + finished.rethrow(); + } + private: std::vector> mReceivers; }; diff --git a/libs/filterGraph/core/filterGraph/FilterGraph.hpp b/libs/filterGraph/core/filterGraph/FilterGraph.hpp index 7ff2ca6..ed807ba 100644 --- a/libs/filterGraph/core/filterGraph/FilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/FilterGraph.hpp @@ -46,6 +46,11 @@ class FilterGraph : public MessageFilter::setContext(std::move(context)); } + void finish() override + { + mFilter->finish(); + } + private: std::shared_ptr mFilter; }; @@ -83,6 +88,14 @@ class FilterGraph : public MessageFilter::setContext(std::move(context)); } + void finish() override + { + detail::FinishScope finished; + finished.run([this] { mFilter->finish(); }); + finished.run([this] { mNext.finish(); }); + finished.rethrow(); + } + private: std::shared_ptr mFilter; FilterGraph mNext; diff --git a/libs/filterGraph/core/filterGraph/JoinFilter.hpp b/libs/filterGraph/core/filterGraph/JoinFilter.hpp index a81adec..557f3e8 100644 --- a/libs/filterGraph/core/filterGraph/JoinFilter.hpp +++ b/libs/filterGraph/core/filterGraph/JoinFilter.hpp @@ -73,6 +73,16 @@ class JoinFilter : public MessageFilter MessageFilter::setContext(std::move(context)); } + void finish() override + { + detail::FinishScope finished; + for (auto& path : mPaths) + { + finished.run([&path] { path->finish(); }); + } + finished.rethrow(); + } + std::optional filter(InputType&& data) override { std::vector gathered; diff --git a/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp b/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp index 4b2d182..c36d155 100644 --- a/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp @@ -68,6 +68,11 @@ class JsonFilterGraph : public MessageFilter MessageFilter::setContext(std::move(context)); } + void finish() override + { + mChain.finish(); + } + private: AnyFilterChain mChain; }; diff --git a/libs/filterGraph/core/filterGraph/MessageFilter.hpp b/libs/filterGraph/core/filterGraph/MessageFilter.hpp index 3f67b45..b3f2410 100644 --- a/libs/filterGraph/core/filterGraph/MessageFilter.hpp +++ b/libs/filterGraph/core/filterGraph/MessageFilter.hpp @@ -2,6 +2,7 @@ #include +#include #include #include #include @@ -9,6 +10,44 @@ namespace filterGraph { +namespace detail { + +// Finishes a group of stages: every stage is finished even if an earlier one +// throws, and the first exception is rethrown afterwards. Composite stages use +// it so that one failing stage cannot keep the others from flushing. +class FinishScope +{ +public: + template + void run(Fn&& fn) + { + try + { + std::forward(fn)(); + } + catch (...) + { + if (!mFirstError) + { + mFirstError = std::current_exception(); + } + } + } + + void rethrow() const + { + if (mFirstError) + { + std::rethrow_exception(mFirstError); + } + } + +private: + std::exception_ptr mFirstError; +}; + +} // namespace detail + // Base interface for a single processing stage. A MessageFilter consumes an // InputType (by rvalue reference, so it may move from / mutate it freely) and // produces an std::optional: returning std::nullopt allows a @@ -31,6 +70,18 @@ class MessageFilter virtual ~MessageFilter() = default; virtual std::optional filter(InputType&& input) = 0; + // Called once after the last message, so that a stage holding state can + // flush it: write a report, close a file, publish a result to the + // GraphContext. It produces no message, so nothing is routed downstream; + // stages that need to emit a final value should publish it through the + // context or their own side channel. + // + // A graph calls finish() on its stages when the graph itself is finished. + // Composite stages override this to forward it to their inner stages. + // Calling it is the owner's decision: a graph that is never finished simply + // never flushes, and finishing twice finishes every stage twice. + virtual void finish() {} + // Composite stages override this to forward the context to their inner // stages. Not meant to be called while messages are being processed. // Passing nullptr installs a fresh empty context rather than none, so the diff --git a/tests/FilterGraphTests/LifecycleTests.cpp b/tests/FilterGraphTests/LifecycleTests.cpp new file mode 100644 index 0000000..b18c77f --- /dev/null +++ b/tests/FilterGraphTests/LifecycleTests.cpp @@ -0,0 +1,194 @@ +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace filterGraph; + +namespace { + +// Every stage appends to this log, so a test can check what ran and in which +// order. Each test clears it first. +std::vector& lifecycleLog() +{ + static std::vector log; + return log; +} + +// Sums the values it sees and writes the total to the log when it is finished: +// the pattern finish() exists for. +class SumFilter : public MessageFilter +{ +public: + explicit SumFilter(std::string label) + : mLabel(std::move(label)) + { + } + + std::optional filter(int&& value) override + { + mSum += value; + return value; + } + + void finish() override + { + lifecycleLog().push_back(std::format("{}={}", mLabel, mSum)); + } + +private: + std::string mLabel; + int mSum = 0; +}; + +// A stage that does not override finish(): the default is a no-op. +class PassThroughFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return value; + } +}; + +class ThrowOnFinishFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return value; + } + + void finish() override + { + lifecycleLog().push_back("throwing"); + throw std::runtime_error("finish failed"); + } +}; + +class ToStringFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return std::to_string(value); + } +}; + +static FilterRegistrar registerPass("LifePass"); +static FilterRegistrar registerToString("LifeToString"); +static FilterRegistrar registerThrow("LifeThrow"); + +static FilterRegistrar registerSum("LifeSum", [](const nlohmann::json& config) { + return std::make_shared(config.value("label", std::string{"sum"})); +}); + +// A nested DSL graph, registered as a stage. +static FilterRegistrar> registerNested("LifeNested", [](const nlohmann::json&) { + return std::make_shared>("in -> LifeSum(label=inner) -> out"); +}); + +} // namespace + +TEST_CASE("DslFilterGraph finishes every stage in run order", "[Lifecycle]") +{ + lifecycleLog().clear(); + DslFilterGraph graph(R"( + in -> LifeSum(label=first) -> a + a -> LifePass -> b + b -> LifeSum(label=second) -> out + )"); + + graph.filter(1); + graph.filter(2); + REQUIRE(lifecycleLog().empty()); // nothing happens until the graph is finished + + graph.finish(); + REQUIRE(lifecycleLog() == std::vector{"first=3", "second=3"}); +} + +TEST_CASE("A nested graph forwards finish to its own stages", "[Lifecycle]") +{ + lifecycleLog().clear(); + DslFilterGraph graph("in -> LifeNested -> nested -> LifeSum(label=outer) -> out"); + + graph.filter(4); + graph.finish(); + + REQUIRE(lifecycleLog() == std::vector{"inner=4", "outer=4"}); +} + +TEST_CASE("The compile-time FilterGraph forwards finish", "[Lifecycle]") +{ + lifecycleLog().clear(); + FilterGraph graph(std::make_shared("head"), + std::make_shared(), + std::make_shared("tail")); + + graph.filter(5); + graph.finish(); + + REQUIRE(lifecycleLog() == std::vector{"head=5", "tail=5"}); +} + +TEST_CASE("JsonFilterGraph forwards finish through fanout branches and join paths", "[Lifecycle]") +{ + lifecycleLog().clear(); + registerFanoutFilter("LifeFanout"); + registerJoinFilter("LifeJoin", [](std::vector&& gathered) { + return std::to_string(gathered.size()); + }); + + JsonFilterGraph graph(nlohmann::json::parse(R"([ + { "type": "LifeFanout", "config": { "branches": [ + [ { "type": "LifeSum", "config": { "label": "branch" } } ] + ] } }, + { "type": "LifeSum", "config": { "label": "main" } }, + { "type": "LifeJoin", "config": { "paths": [ + [ { "type": "LifeSum", "config": { "label": "path" } }, { "type": "LifeToString" } ] + ] } } + ])")); + + graph.filter(6); + graph.finish(); + + REQUIRE(lifecycleLog() == std::vector{"branch=6", "main=6", "path=6"}); +} + +TEST_CASE("Every stage is finished even when one throws", "[Lifecycle]") +{ + lifecycleLog().clear(); + DslFilterGraph graph(R"( + in -> LifeThrow -> a + a -> LifeSum(label=after) -> out + )"); + + graph.filter(7); + REQUIRE_THROWS_AS(graph.finish(), std::runtime_error); + REQUIRE(lifecycleLog() == std::vector{"throwing", "after=7"}); +} + +TEST_CASE("Finishing a graph twice finishes its stages twice", "[Lifecycle]") +{ + lifecycleLog().clear(); + DslFilterGraph graph("in -> LifeSum(label=once) -> out"); + + graph.filter(8); + graph.finish(); + graph.finish(); + + REQUIRE(lifecycleLog() == std::vector{"once=8", "once=8"}); +}