From f4b65f059f96bd7240e75c4016eb36890d9e3ca3 Mon Sep 17 00:00:00 2001 From: psy_inf Date: Thu, 17 Sep 2026 19:19:50 +0200 Subject: [PATCH 1/2] feat: add GraphContext, a graph-scoped side channel for stages - GraphContext: type-keyed, thread-safe blackboard (set / get / getOr / contains / erase / update / clear); values are copied in and out under a shared_mutex, update() is atomic - polymorphic base: derive an application context and recover it with as() - MessageFilter::setContext / context(): a graph hands its context to its stages once; FilterGraph, DslFilterGraph, JsonFilterGraph, AnyFilterChain, FanoutFilter and JoinFilter forward it, including to nested graphs and to receivers/paths added later - breaking: AnyMessageFilter::setContext is pure virtual - README section and CHANGELOG entries --- CHANGELOG.md | 10 + README.md | 58 ++++ libs/filterGraph/CMakeLists.txt | 1 + .../core/filterGraph/AnyFilterChain.hpp | 8 + .../core/filterGraph/AnyMessageFilter.hpp | 9 + .../core/filterGraph/DslFilterGraph.hpp | 14 + .../core/filterGraph/FanoutFilter.hpp | 10 + .../core/filterGraph/FilterGraph.hpp | 13 + .../core/filterGraph/GraphContext.hpp | 143 +++++++++ .../core/filterGraph/JoinFilter.hpp | 10 + .../core/filterGraph/JsonFilterGraph.hpp | 6 + .../core/filterGraph/MessageFilter.hpp | 23 ++ tests/FilterGraphTests/GraphContextTests.cpp | 284 ++++++++++++++++++ 13 files changed, 589 insertions(+) create mode 100644 libs/filterGraph/core/filterGraph/GraphContext.hpp create mode 100644 tests/FilterGraphTests/GraphContextTests.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index 75e5c35..82562a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,8 +23,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`validateDslGraph(text)`** — every build-time diagnostic without throwing; `dsl::formatDiagnostic(s)` formats them as `line:column: message`. - `dsl::StageNode::fanIn` records whether a stage's inputs came from a group. +- **`GraphContext`** (`GraphContext.hpp`) — a graph-scoped, type-keyed, + thread-safe blackboard for side-channel data between stages + (`set` / `get` / `getOr` / `contains` / `erase` / `update` / `clear`). It is a + polymorphic base; derived contexts are recovered with `as()`. +- **`MessageFilter::setContext` / `context()`** — stages receive the graph's + `GraphContext`. `FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`, + `AnyFilterChain`, `FanoutFilter` and `JoinFilter` forward it to their stages. ### Changed +- **Breaking:** `AnyMessageFilter` has a new pure virtual + `setContext(std::shared_ptr)`; custom implementations must + forward the context to the stages they wrap. - 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 4fba902..5b4ac44 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,9 @@ drops the message, and everything downstream of that edge is skipped. runtime configuration overhead. - **`SinkFilter`** — a generic terminal stage that forwards data to a caller-supplied `std::function` callback. +- **`GraphContext`** — a graph-scoped, type-keyed, thread-safe blackboard, so + stages can share values without knowing who produced them. See + [Graph context](#graph-context). ### Runtime graphs in the text DSL @@ -317,6 +320,61 @@ listing each stage's arguments. C++ template parameters; the types inside `GraphOutputs` are checked when they are read (`get` throws `std::bad_any_cast` on a mismatch). +## Graph context + +`GraphContext` (`GraphContext.hpp`) is a side channel for data that is not part +of the message, such as a frame number published by one stage and read by +another. The value's type is the key, so there is at most one value per type: + +```cpp +struct FrameNo { std::uint64_t value; }; // wrap primitives in a dedicated type + +auto ctx = std::make_shared(); +ctx->set(FrameNo{42}); // publish (replaces) +std::optional frame = ctx->get(); // nullopt if unset +FrameNo orDefault = ctx->getOr(FrameNo{0}); +ctx->update([](FrameNo& f) { ++f.value; }); // atomic; false if unset +``` + +- Values are copied in and out under an internal lock, so a context can be + shared by concurrently running paths. Store `std::shared_ptr` for heavy or + non-copyable data. The callback given to `update` must not access the context. +- `GraphContext` is a polymorphic base: derive an application context, pass it + around as `std::shared_ptr`, and recover it with + `ctx->as()` (`nullptr` if it is another type). Members added by a + derived class are not covered by the lock, and a stage calling `as<>` depends + on that type, so reusable stages should stick to `set` / `get`. +- The context is graph-scoped, not per message: if the graph ever buffers or + reorders messages, a value such as a frame number may belong to another + message than the one being processed. + +Hand the context to a graph once, before processing messages. Every graph type +(`FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`) and composite stage +(`FanoutFilter`, `JoinFilter`, nested graphs) forwards it to its stages, which +read it through `MessageFilter::context()`: + +```cpp +class StampFrame : public MessageFilter +{ +public: + std::optional filter(Image&& image) override + { + if (context()) // nullptr outside a graph or when no context was set + { + image.frame = context()->getOr(FrameNo{0}).value; + } + return std::move(image); + } +}; + +DslFilterGraph graph(text); +graph.setContext(ctx); +``` + +`setContext` is not meant to be called while messages are being processed. +Custom composite stages override it to forward the context to their inner +stages. + ## JSON format The JSON format predates the DSL and remains fully supported; both use the diff --git a/libs/filterGraph/CMakeLists.txt b/libs/filterGraph/CMakeLists.txt index 7f6d395..8cb01a4 100644 --- a/libs/filterGraph/CMakeLists.txt +++ b/libs/filterGraph/CMakeLists.txt @@ -20,6 +20,7 @@ target_sources(${PROJECT_NAME} core/filterGraph/GraphLangLexy.hpp core/filterGraph/GraphLangHandwritten.hpp core/filterGraph/DslFilterGraph.hpp + core/filterGraph/GraphContext.hpp ) target_link_libraries(${PROJECT_NAME} diff --git a/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp b/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp index 0baada5..c01ad0a 100644 --- a/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp +++ b/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp @@ -92,6 +92,14 @@ class AnyFilterChain : public AnyMessageFilter return mStages.back()->outputType(); } + void setContext(std::shared_ptr context) override + { + for (auto& stage : mStages) + { + stage->setContext(context); + } + } + private: std::vector> mStages; }; diff --git a/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp b/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp index d02333f..ec233ac 100644 --- a/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp +++ b/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp @@ -26,6 +26,10 @@ class AnyMessageFilter virtual std::type_index inputType() const = 0; virtual std::type_index outputType() const = 0; + + // Hands the graph's context to the wrapped stage(s); see + // MessageFilter::setContext. + virtual void setContext(std::shared_ptr context) = 0; }; // Adapts a concrete MessageFilter to the @@ -60,6 +64,11 @@ class AnyMessageFilterAdapter : public AnyMessageFilter return typeid(typename Filter::OutType); } + void setContext(std::shared_ptr context) override + { + mFilter->setContext(std::move(context)); + } + private: std::shared_ptr mFilter; }; diff --git a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp index 8fb497c..58c05de 100644 --- a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp @@ -431,6 +431,14 @@ class GraphPlan return mOutputKeys; } + void setContext(const std::shared_ptr& context) + { + for (auto& stage : mStages) + { + stage.filter->setContext(context); + } + } + private: static constexpr std::size_t kInputSlot = 0; @@ -575,6 +583,12 @@ class DslFilterGraph : public MessageFilter } } + void setContext(std::shared_ptr context) override + { + mPlan.setContext(context); + MessageFilter::setContext(std::move(context)); + } + // 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 2959137..70da9fd 100644 --- a/libs/filterGraph/core/filterGraph/FanoutFilter.hpp +++ b/libs/filterGraph/core/filterGraph/FanoutFilter.hpp @@ -43,6 +43,7 @@ class FanoutFilter : public MessageFilter receiver->inputType().name(), typeid(InputType).name())); } + receiver->setContext(this->context()); mReceivers.push_back(std::move(receiver)); } @@ -56,6 +57,15 @@ class FanoutFilter : public MessageFilter return std::move(data); } + void setContext(std::shared_ptr context) override + { + for (auto& receiver : mReceivers) + { + receiver->setContext(context); + } + MessageFilter::setContext(std::move(context)); + } + private: std::vector> mReceivers; }; diff --git a/libs/filterGraph/core/filterGraph/FilterGraph.hpp b/libs/filterGraph/core/filterGraph/FilterGraph.hpp index b124237..e3cc7ea 100644 --- a/libs/filterGraph/core/filterGraph/FilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/FilterGraph.hpp @@ -39,6 +39,12 @@ class FilterGraph : public MessageFilterfilter(std::move(data)); } + void setContext(std::shared_ptr context) override + { + mFilter->setContext(context); + MessageFilter::setContext(std::move(context)); + } + private: std::shared_ptr mFilter; }; @@ -68,6 +74,13 @@ class FilterGraph : public MessageFilter context) override + { + mFilter->setContext(context); + mNext.setContext(context); + MessageFilter::setContext(std::move(context)); + } + private: std::shared_ptr mFilter; FilterGraph mNext; diff --git a/libs/filterGraph/core/filterGraph/GraphContext.hpp b/libs/filterGraph/core/filterGraph/GraphContext.hpp new file mode 100644 index 0000000..728a12a --- /dev/null +++ b/libs/filterGraph/core/filterGraph/GraphContext.hpp @@ -0,0 +1,143 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace filterGraph { + +// A value stored in a GraphContext: a plain (non-cv, non-reference) copyable +// type. The type itself is the key, so wrap primitives in a dedicated struct +// (struct FrameNo { std::uint64_t value; };) instead of storing a bare int +// that unrelated stages would silently share. Non-copyable or heavy payloads +// can be stored as std::shared_ptr. +template +concept ContextValue = std::same_as> && std::copy_constructible; + +class GraphContext; + +template +concept ContextType = std::derived_from; + +// GraphContext is a graph-scoped, type-keyed blackboard: any stage may publish +// a value and any other stage may read it without knowing who wrote it. There +// is at most one value per type. +// +// Values are copied in and out, so no reference to stored state escapes the +// internal lock; this keeps the context safe to share between concurrently +// running paths. Use update() for an atomic read-modify-write. +// +// GraphContext is a polymorphic base: an application may derive its own +// context with dedicated members, hand it to the graph as a GraphContext, and +// recover it in stages via as(). Members added by a derived class are +// not covered by the internal lock. Prefer the type-keyed store for stages +// meant to be reusable, as as() couples a stage to that type. +class GraphContext +{ +public: + GraphContext() = default; + GraphContext(const GraphContext&) = delete; + GraphContext& operator=(const GraphContext&) = delete; + virtual ~GraphContext() = default; + + // Returns this context as Derived, or nullptr if it is not one. + template + [[nodiscard]] Derived* as() noexcept + { + return dynamic_cast(this); + } + + template + [[nodiscard]] const Derived* as() const noexcept + { + return dynamic_cast(this); + } + + // Stores value, replacing any previous value of the same type. + template + void set(T value) + { + std::unique_lock lock(mMutex); + mValues.insert_or_assign(std::type_index(typeid(T)), std::any(std::move(value))); + } + + // Returns a copy of the stored value, or std::nullopt if none was set. + template + [[nodiscard]] std::optional get() const + { + std::shared_lock lock(mMutex); + if (const auto* value = findLocked()) + { + return *value; + } + return std::nullopt; + } + + // Returns a copy of the stored value, or fallback if none was set. + template + [[nodiscard]] T getOr(T fallback) const + { + std::shared_lock lock(mMutex); + if (const auto* value = findLocked()) + { + return *value; + } + return fallback; + } + + template + [[nodiscard]] bool contains() const + { + std::shared_lock lock(mMutex); + return mValues.contains(std::type_index(typeid(T))); + } + + // Removes the value of type T; returns whether one was present. + template + bool erase() + { + std::unique_lock lock(mMutex); + return mValues.erase(std::type_index(typeid(T))) > 0; + } + + // Atomically modifies an existing value in place; returns false (without + // calling fn) if no value of type T is present. fn runs under the context's + // lock and must not access the context itself. + template Fn> + bool update(Fn&& fn) + { + std::unique_lock lock(mMutex); + auto it = mValues.find(std::type_index(typeid(T))); + if (it == mValues.end()) + { + return false; + } + std::invoke(std::forward(fn), *std::any_cast(&it->second)); + return true; + } + + void clear() + { + std::unique_lock lock(mMutex); + mValues.clear(); + } + +private: + template + const T* findLocked() const + { + auto it = mValues.find(std::type_index(typeid(T))); + return it == mValues.end() ? nullptr : std::any_cast(&it->second); + } + + mutable std::shared_mutex mMutex; + std::unordered_map mValues; +}; + +} // namespace filterGraph diff --git a/libs/filterGraph/core/filterGraph/JoinFilter.hpp b/libs/filterGraph/core/filterGraph/JoinFilter.hpp index 1b6cb87..f862488 100644 --- a/libs/filterGraph/core/filterGraph/JoinFilter.hpp +++ b/libs/filterGraph/core/filterGraph/JoinFilter.hpp @@ -60,9 +60,19 @@ class JoinFilter : public MessageFilter throw std::runtime_error( "JoinFilter: a Void-terminated path produces no value to join; join paths must produce a value"); } + path->setContext(this->context()); mPaths.push_back(std::move(path)); } + void setContext(std::shared_ptr context) override + { + for (auto& path : mPaths) + { + path->setContext(context); + } + MessageFilter::setContext(std::move(context)); + } + 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 1abe851..1786bba 100644 --- a/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp @@ -60,6 +60,12 @@ class JsonFilterGraph : public MessageFilter return std::any_cast(std::move(*result)); } + void setContext(std::shared_ptr context) override + { + mChain.setContext(context); + MessageFilter::setContext(std::move(context)); + } + private: AnyFilterChain mChain; }; diff --git a/libs/filterGraph/core/filterGraph/MessageFilter.hpp b/libs/filterGraph/core/filterGraph/MessageFilter.hpp index 0ad6673..9499d53 100644 --- a/libs/filterGraph/core/filterGraph/MessageFilter.hpp +++ b/libs/filterGraph/core/filterGraph/MessageFilter.hpp @@ -1,6 +1,9 @@ #pragma once +#include + #include +#include #include #include @@ -11,6 +14,11 @@ namespace filterGraph { // produces an std::optional: returning std::nullopt allows a // stage to short-circuit (terminate) a chain early, e.g. when a message // should be dropped/filtered out. +// +// A stage may read and publish side-channel data through context(), the +// GraphContext of the graph it runs in. The graph hands it over once via +// setContext() before messages are processed; a stage used outside a graph, or +// in a graph without a context, sees nullptr. template class MessageFilter { @@ -20,6 +28,21 @@ class MessageFilter virtual ~MessageFilter() = default; virtual std::optional filter(InputType&& input) = 0; + + // Composite stages override this to forward the context to their inner + // stages. Not meant to be called while messages are being processed. + virtual void setContext(std::shared_ptr context) + { + mContext = std::move(context); + } + + [[nodiscard]] const std::shared_ptr& context() const noexcept + { + return mContext; + } + +private: + std::shared_ptr mContext; }; // Generic terminal stage for a FilterGraph. Consumes InputType via a caller diff --git a/tests/FilterGraphTests/GraphContextTests.cpp b/tests/FilterGraphTests/GraphContextTests.cpp new file mode 100644 index 0000000..f0853b3 --- /dev/null +++ b/tests/FilterGraphTests/GraphContextTests.cpp @@ -0,0 +1,284 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace filterGraph; + +namespace { + +struct FrameNo +{ + std::uint64_t value{}; +}; + +struct SourceName +{ + std::string value; +}; + +struct Counter +{ + int value{}; +}; + +class AppContext : public GraphContext +{ +public: + std::string sessionId = "session-1"; +}; + +class OtherContext : public GraphContext +{ +}; + +struct SeenFrame +{ + std::uint64_t value{}; +}; + +// Publishes each message as the current frame number. +class PublishFrameFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + context()->set(FrameNo{static_cast(value)}); + return value; + } +}; + +// Replaces the message by the published frame number, or -1 without a context. +class ReadFrameFilter : public MessageFilter +{ +public: + std::optional filter(int&&) override + { + if (!context()) + { + return -1; + } + return static_cast(context()->getOr(FrameNo{}).value); + } +}; + +// Records the frame number it saw, for side branches whose output is discarded. +class RecordFrameFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + context()->set(SeenFrame{context()->getOr(FrameNo{}).value}); + return value; + } +}; + +std::shared_ptr erase(std::shared_ptr> filter) +{ + return std::make_shared>>(std::move(filter)); +} + +} // namespace + +TEST_CASE("A stage outside a graph has no context", "[GraphContext]") +{ + ReadFrameFilter read; + REQUIRE(read.context() == nullptr); + REQUIRE(read.filter(5) == -1); +} + +TEST_CASE("FilterGraph hands its context to every stage", "[GraphContext]") +{ + FilterGraph graph( + std::make_shared(), std::make_shared()); + + auto ctx = std::make_shared(); + graph.setContext(ctx); + + REQUIRE(graph.context() == ctx); + REQUIRE(graph.filter(5) == 5); + REQUIRE(ctx->get()->value == 5); +} + +TEST_CASE("DslFilterGraph lets stages exchange values through the context", "[GraphContext]") +{ + static FilterRegistrar registerPublish("CtxPublish"); + static FilterRegistrar registerRead("CtxRead"); + + DslFilterGraph graph(R"( + in -> CtxPublish -> published + published -> CtxRead -> out + )"); + + auto ctx = std::make_shared(); + graph.setContext(ctx); + + REQUIRE(graph.filter(7) == 7); + REQUIRE(graph.filter(8) == 8); + REQUIRE(ctx->get()->value == 8); +} + +TEST_CASE("Nested JSON fanout and join paths receive the context", "[GraphContext]") +{ + static FilterRegistrar registerPublish("CtxJsonPublish"); + static FilterRegistrar registerRead("CtxJsonRead"); + static FilterRegistrar registerRecord("CtxJsonRecord"); + registerFanoutFilter("CtxJsonFanout"); + registerJoinFilter("CtxJsonJoin", [](std::vector&& slots) -> std::optional { + return std::any_cast(slots.at(0)); + }); + + JsonFilterGraph graph(nlohmann::json::parse(R"([ + { "type": "CtxJsonPublish" }, + { "type": "CtxJsonFanout", "config": { "branches": [ [ { "type": "CtxJsonRecord" } ] ] } }, + { "type": "CtxJsonJoin", "config": { "paths": [ [ { "type": "CtxJsonRead" } ] ] } } + ])")); + + auto ctx = std::make_shared(); + graph.setContext(ctx); + + REQUIRE(graph.filter(11) == 11); + REQUIRE(ctx->get()->value == 11); +} + +TEST_CASE("A receiver added after setContext still receives the context", "[GraphContext]") +{ + auto ctx = std::make_shared(); + ctx->set(FrameNo{9}); + + FanoutFilter fanout; + fanout.setContext(ctx); + fanout.addReceiver(erase(std::make_shared())); + + REQUIRE(fanout.filter(3) == 3); + REQUIRE(ctx->get()->value == 9); +} + +TEST_CASE("GraphContext can be extended and recovered polymorphically", "[GraphContext]") +{ + std::shared_ptr ctx = std::make_shared(); + ctx->set(FrameNo{5}); + + REQUIRE(ctx->as() != nullptr); + REQUIRE(ctx->as()->sessionId == "session-1"); + REQUIRE(ctx->as() == nullptr); + REQUIRE(ctx->as()->get()->value == 5); + + std::shared_ptr constCtx = ctx; + REQUIRE(constCtx->as() != nullptr); +} + +TEST_CASE("GraphContext returns nothing for unset types", "[GraphContext]") +{ + GraphContext ctx; + + REQUIRE_FALSE(ctx.contains()); + REQUIRE_FALSE(ctx.get().has_value()); + REQUIRE(ctx.getOr(FrameNo{7}).value == 7); + REQUIRE_FALSE(ctx.erase()); +} + +TEST_CASE("GraphContext stores one value per type", "[GraphContext]") +{ + GraphContext ctx; + ctx.set(FrameNo{42}); + ctx.set(SourceName{"camera"}); + + REQUIRE(ctx.contains()); + REQUIRE(ctx.get()->value == 42); + REQUIRE(ctx.get()->value == "camera"); + + ctx.set(FrameNo{43}); + REQUIRE(ctx.get()->value == 43); + REQUIRE(ctx.getOr(FrameNo{0}).value == 43); +} + +TEST_CASE("GraphContext hands out copies, not references", "[GraphContext]") +{ + GraphContext ctx; + ctx.set(SourceName{"camera"}); + + auto copy = *ctx.get(); + copy.value = "changed"; + + REQUIRE(ctx.get()->value == "camera"); +} + +TEST_CASE("GraphContext erase and clear remove values", "[GraphContext]") +{ + GraphContext ctx; + ctx.set(FrameNo{1}); + ctx.set(SourceName{"camera"}); + + REQUIRE(ctx.erase()); + REQUIRE_FALSE(ctx.contains()); + REQUIRE(ctx.contains()); + + ctx.clear(); + REQUIRE_FALSE(ctx.contains()); +} + +TEST_CASE("GraphContext update modifies existing values only", "[GraphContext]") +{ + GraphContext ctx; + + bool called = false; + REQUIRE_FALSE(ctx.update([&](Counter&) { called = true; })); + REQUIRE_FALSE(called); + + ctx.set(Counter{1}); + REQUIRE(ctx.update([](Counter& counter) { ++counter.value; })); + REQUIRE(ctx.get()->value == 2); +} + +TEST_CASE("GraphContext shares non-copyable payloads via shared_ptr", "[GraphContext]") +{ + GraphContext ctx; + ctx.set(std::make_shared>(std::vector{1, 2, 3})); + + auto shared = ctx.get>>(); + REQUIRE(shared.has_value()); + (*shared)->push_back(4); + + REQUIRE((*ctx.get>>())->size() == 4); +} + +TEST_CASE("GraphContext update is atomic across threads", "[GraphContext]") +{ + GraphContext ctx; + ctx.set(Counter{0}); + + constexpr int threads = 8; + constexpr int increments = 1000; + + std::vector workers; + for (int t = 0; t < threads; ++t) + { + workers.emplace_back([&] { + for (int i = 0; i < increments; ++i) + { + ctx.update([](Counter& counter) { ++counter.value; }); + } + }); + } + workers.clear(); + + REQUIRE(ctx.get()->value == threads * increments); +} From 4094da8020a2d43c960e3ffaa9f78b905d2496b3 Mon Sep 17 00:00:00 2001 From: psy_inf Date: Thu, 17 Sep 2026 20:12:40 +0200 Subject: [PATCH 2/2] feat: every stage always has a GraphContext context() is never null: a stage starts with its own empty context, and every graph hands its context to its stages when it is built, not only on an explicit setContext. So stages of a graph share one context out of the box and no stage needs a null check. - MessageFilter: mContext is always set; setContext(nullptr) installs a fresh empty context; context() returns GraphContext& (const overload too) and sharedContext() the shared_ptr for composites to forward - FilterGraph, JsonFilterGraph, DslFilterGraph propagate their context from their constructor; AnyFilterChain owns one and does the same, so a chain built on its own has one context, not one per stage - FanoutFilter / JoinFilter hand sharedContext() to late receivers/paths - a graph overwrites the context of the stages it is built from, so a shared context goes to the graph, not to individual stages --- CHANGELOG.md | 11 +++-- README.md | 37 ++++++++++---- .../core/filterGraph/AnyFilterChain.hpp | 8 ++- .../core/filterGraph/DslFilterGraph.hpp | 2 + .../core/filterGraph/FanoutFilter.hpp | 2 +- .../core/filterGraph/FilterGraph.hpp | 2 + .../core/filterGraph/JoinFilter.hpp | 2 +- .../core/filterGraph/JsonFilterGraph.hpp | 2 + .../core/filterGraph/MessageFilter.hpp | 29 ++++++++--- tests/FilterGraphTests/GraphContextTests.cpp | 49 ++++++++++++++----- 10 files changed, 110 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 82562a9..889681b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,9 +27,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 thread-safe blackboard for side-channel data between stages (`set` / `get` / `getOr` / `contains` / `erase` / `update` / `clear`). It is a polymorphic base; derived contexts are recovered with `as()`. -- **`MessageFilter::setContext` / `context()`** — stages receive the graph's - `GraphContext`. `FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`, - `AnyFilterChain`, `FanoutFilter` and `JoinFilter` forward it to their stages. +- **`MessageFilter::setContext` / `context()` / `sharedContext()`** — stages + receive the graph's `GraphContext`. `FilterGraph`, `DslFilterGraph`, + `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter` and `JoinFilter` create an + empty context when they are built and forward it to their stages, so stages of + a graph share one context without any setup; `setContext` replaces it (with + nullptr meaning "a fresh empty one"). `context()` returns a `GraphContext&` + and is never null, so stages need no null check; `sharedContext()` hands out + the `std::shared_ptr` for composites that forward it. ### Changed - **Breaking:** `AnyMessageFilter` has a new pure virtual diff --git a/README.md b/README.md index 5b4ac44..384b8cc 100644 --- a/README.md +++ b/README.md @@ -348,10 +348,8 @@ ctx->update([](FrameNo& f) { ++f.value; }); // atomic; false if u reorders messages, a value such as a frame number may belong to another message than the one being processed. -Hand the context to a graph once, before processing messages. Every graph type -(`FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`) and composite stage -(`FanoutFilter`, `JoinFilter`, nested graphs) forwards it to its stages, which -read it through `MessageFilter::context()`: +There is always a context, so a stage never has to check for one. A stage +reads it through `MessageFilter::context()`, which returns a `GraphContext&`: ```cpp class StampFrame : public MessageFilter @@ -359,21 +357,40 @@ class StampFrame : public MessageFilter public: std::optional filter(Image&& image) override { - if (context()) // nullptr outside a graph or when no context was set - { - image.frame = context()->getOr(FrameNo{0}).value; - } + image.frame = context().getOr(FrameNo{0}).value; return std::move(image); } }; +``` + +Every graph type (`FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`) creates +an empty context when it is built and hands it to its stages, and so does every +composite stage (`FanoutFilter`, `JoinFilter`, nested graphs), including to +receivers and paths added later. So stages of the same graph share one context +out of the box: +```cpp DslFilterGraph graph(text); -graph.setContext(ctx); +graph.context().set(FrameNo{1}); // the graph's own context +``` + +Call `setContext` to put a different context in its place — typically an +application's derived one, or a single context shared by several graphs: + +```cpp +graph.setContext(ctx); // replaces the graph's own context ``` +A stage that is not part of a graph keeps its own empty context, which nobody +else sees: it works, but nothing is shared. Note that the graph a stage is +added to overwrites the stage's context with its own, so hand a shared context +to the graph rather than to individual stages. `setContext(nullptr)` installs a +fresh empty context rather than none. + `setContext` is not meant to be called while messages are being processed. Custom composite stages override it to forward the context to their inner -stages. +stages; `MessageFilter::sharedContext()` returns the `std::shared_ptr` to pass +on. ## JSON format diff --git a/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp b/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp index c01ad0a..b654a3b 100644 --- a/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp +++ b/libs/filterGraph/core/filterGraph/AnyFilterChain.hpp @@ -61,6 +61,8 @@ class AnyFilterChain : public AnyMessageFilter mStages.push_back(std::move(stage)); } + + AnyFilterChain::setContext(mContext); } std::optional filter(std::any&& input) override @@ -94,14 +96,18 @@ class AnyFilterChain : public AnyMessageFilter void setContext(std::shared_ptr context) override { + mContext = context ? std::move(context) : std::make_shared(); for (auto& stage : mStages) { - stage->setContext(context); + stage->setContext(mContext); } } private: std::vector> mStages; + // Shared by the stages of this chain until a graph hands down its own, so + // that a chain built on its own still has one context, not one per stage. + std::shared_ptr mContext = std::make_shared(); }; } // namespace filterGraph diff --git a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp index 58c05de..7d73bb1 100644 --- a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp @@ -552,6 +552,8 @@ class DslFilterGraph : public MessageFilter dsl::detail::sortDiagnostics(diagnostics); throw GraphError(std::move(diagnostics)); } + + DslFilterGraph::setContext(this->sharedContext()); } std::optional filter(InputType&& data) override diff --git a/libs/filterGraph/core/filterGraph/FanoutFilter.hpp b/libs/filterGraph/core/filterGraph/FanoutFilter.hpp index 70da9fd..b34c4c1 100644 --- a/libs/filterGraph/core/filterGraph/FanoutFilter.hpp +++ b/libs/filterGraph/core/filterGraph/FanoutFilter.hpp @@ -43,7 +43,7 @@ class FanoutFilter : public MessageFilter receiver->inputType().name(), typeid(InputType).name())); } - receiver->setContext(this->context()); + receiver->setContext(this->sharedContext()); mReceivers.push_back(std::move(receiver)); } diff --git a/libs/filterGraph/core/filterGraph/FilterGraph.hpp b/libs/filterGraph/core/filterGraph/FilterGraph.hpp index e3cc7ea..7ff2ca6 100644 --- a/libs/filterGraph/core/filterGraph/FilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/FilterGraph.hpp @@ -32,6 +32,7 @@ class FilterGraph : public MessageFilter filter) : mFilter(std::move(filter)) { + FilterGraph::setContext(this->sharedContext()); } std::optional filter(InType&& data) override @@ -62,6 +63,7 @@ class FilterGraph : public MessageFiltersharedContext()); } std::optional filter(InType&& data) override diff --git a/libs/filterGraph/core/filterGraph/JoinFilter.hpp b/libs/filterGraph/core/filterGraph/JoinFilter.hpp index f862488..a81adec 100644 --- a/libs/filterGraph/core/filterGraph/JoinFilter.hpp +++ b/libs/filterGraph/core/filterGraph/JoinFilter.hpp @@ -60,7 +60,7 @@ class JoinFilter : public MessageFilter throw std::runtime_error( "JoinFilter: a Void-terminated path produces no value to join; join paths must produce a value"); } - path->setContext(this->context()); + path->setContext(this->sharedContext()); mPaths.push_back(std::move(path)); } diff --git a/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp b/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp index 1786bba..4b2d182 100644 --- a/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp @@ -48,6 +48,8 @@ class JsonFilterGraph : public MessageFilter mChain.outputType().name(), typeid(OutputType).name())); } + + JsonFilterGraph::setContext(this->sharedContext()); } std::optional filter(InputType&& data) override diff --git a/libs/filterGraph/core/filterGraph/MessageFilter.hpp b/libs/filterGraph/core/filterGraph/MessageFilter.hpp index 9499d53..3f67b45 100644 --- a/libs/filterGraph/core/filterGraph/MessageFilter.hpp +++ b/libs/filterGraph/core/filterGraph/MessageFilter.hpp @@ -16,9 +16,11 @@ namespace filterGraph { // should be dropped/filtered out. // // A stage may read and publish side-channel data through context(), the -// GraphContext of the graph it runs in. The graph hands it over once via -// setContext() before messages are processed; a stage used outside a graph, or -// in a graph without a context, sees nullptr. +// GraphContext of the graph it runs in. A context always exists: every stage +// starts with its own empty one, every graph hands its context to its stages +// when it is built, and setContext() replaces it (e.g. with an application's +// derived context). So context() never needs a null check; a stage used +// outside a graph simply talks to a context nobody else sees. template class MessageFilter { @@ -31,18 +33,33 @@ class MessageFilter // 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 + // "there is always a context" invariant holds unconditionally. virtual void setContext(std::shared_ptr context) { - mContext = std::move(context); + mContext = context ? std::move(context) : std::make_shared(); } - [[nodiscard]] const std::shared_ptr& context() const noexcept + // The context this stage runs in. Never null. + [[nodiscard]] GraphContext& context() noexcept + { + return *mContext; + } + + [[nodiscard]] const GraphContext& context() const noexcept + { + return *mContext; + } + + // The same context as a shared_ptr, for composites handing it to stages + // they own. + [[nodiscard]] const std::shared_ptr& sharedContext() const noexcept { return mContext; } private: - std::shared_ptr mContext; + std::shared_ptr mContext = std::make_shared(); }; // Generic terminal stage for a FilterGraph. Consumes InputType via a caller diff --git a/tests/FilterGraphTests/GraphContextTests.cpp b/tests/FilterGraphTests/GraphContextTests.cpp index f0853b3..a01a996 100644 --- a/tests/FilterGraphTests/GraphContextTests.cpp +++ b/tests/FilterGraphTests/GraphContextTests.cpp @@ -60,22 +60,18 @@ class PublishFrameFilter : public MessageFilter public: std::optional filter(int&& value) override { - context()->set(FrameNo{static_cast(value)}); + context().set(FrameNo{static_cast(value)}); return value; } }; -// Replaces the message by the published frame number, or -1 without a context. +// Replaces the message by the published frame number (0 if none was published). class ReadFrameFilter : public MessageFilter { public: std::optional filter(int&&) override { - if (!context()) - { - return -1; - } - return static_cast(context()->getOr(FrameNo{}).value); + return static_cast(context().getOr(FrameNo{}).value); } }; @@ -85,7 +81,7 @@ class RecordFrameFilter : public MessageFilter public: std::optional filter(int&& value) override { - context()->set(SeenFrame{context()->getOr(FrameNo{}).value}); + context().set(SeenFrame{context().getOr(FrameNo{}).value}); return value; } }; @@ -97,11 +93,40 @@ std::shared_ptr erase(std::shared_ptr> filt } // namespace -TEST_CASE("A stage outside a graph has no context", "[GraphContext]") +TEST_CASE("A stage outside a graph has an empty context of its own", "[GraphContext]") { ReadFrameFilter read; - REQUIRE(read.context() == nullptr); - REQUIRE(read.filter(5) == -1); + REQUIRE(read.sharedContext() != nullptr); + REQUIRE_FALSE(read.context().contains()); + REQUIRE(read.filter(5) == 0); + + read.context().set(FrameNo{4}); + REQUIRE(read.filter(5) == 4); + + ReadFrameFilter other; + REQUIRE(other.sharedContext() != read.sharedContext()); + REQUIRE_FALSE(other.context().contains()); +} + +TEST_CASE("A graph gives its stages a shared context without setContext", "[GraphContext]") +{ + FilterGraph graph( + std::make_shared(), std::make_shared()); + + REQUIRE(graph.sharedContext() != nullptr); + REQUIRE(graph.filter(5) == 5); + REQUIRE(graph.context().get()->value == 5); +} + +TEST_CASE("setContext(nullptr) installs a fresh empty context", "[GraphContext]") +{ + ReadFrameFilter read; + read.context().set(FrameNo{3}); + + read.setContext(nullptr); + + REQUIRE(read.sharedContext() != nullptr); + REQUIRE_FALSE(read.context().contains()); } TEST_CASE("FilterGraph hands its context to every stage", "[GraphContext]") @@ -112,7 +137,7 @@ TEST_CASE("FilterGraph hands its context to every stage", "[GraphContext]") auto ctx = std::make_shared(); graph.setContext(ctx); - REQUIRE(graph.context() == ctx); + REQUIRE(graph.sharedContext() == ctx); REQUIRE(graph.filter(5) == 5); REQUIRE(ctx->get()->value == 5); }