From daa843e6906045f5c849d24e223edf2f19aa4661 Mon Sep 17 00:00:00 2001 From: psy_inf Date: Sat, 19 Sep 2026 21:32:03 +0200 Subject: [PATCH] feat: named merge slots A fan-in group can name the slots of a merge, `(raw: msg, checked: valid) -> Merge`, and is then matched by name, in any order; the merge receives its slots in the order it declares them. This catches two same-typed slots wired the wrong way round, which slot types alone cannot. Merges declare names with MergeStage::mergeInputNames(), or with new registerTypedMergeFilter / registerMergeFilter overloads taking the names. Positional groups keep working. Unknown or missing names, partly named groups and names on a merge without declared names are build-time diagnostics; toMermaid labels the links with the slot names. Implements TODO item 1. --- CHANGELOG.md | 16 + EXAMPLE.md | 44 +++ README.md | 41 ++- TODO.md | 77 ++--- apps/textPipeline/main.cpp | 37 ++- .../core/filterGraph/AnyMessageFilter.hpp | 18 ++ .../core/filterGraph/DslFilterGraph.hpp | 160 ++++++++-- .../core/filterGraph/GraphLang.hpp | 72 ++++- .../core/filterGraph/GraphLangHandwritten.hpp | 24 +- .../core/filterGraph/MergeFilter.hpp | 70 +++- .../core/filterGraph/MergeStage.hpp | 21 +- tests/FilterGraphTests/FilterGraphTests.cpp | 10 + tests/FilterGraphTests/NamedSlotTests.cpp | 299 ++++++++++++++++++ 13 files changed, 785 insertions(+), 104 deletions(-) create mode 100644 tests/FilterGraphTests/NamedSlotTests.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index efcf857..90e8bcd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 stage of an outer graph (with `finish()` and the context reaching into it), a `Void`-terminated sink graph, and the in-band tick message. EXAMPLE.md walks through both in sections 8 and 9. +- **Named merge slots** — a fan-in group can name the slots of a merge, + `(raw: msg, checked: valid) -> Merge`, and is then matched by name, in any + order: the merge receives its slots in the order it declares them. A merge + declares names with `MergeStage::mergeInputNames()` (override it in a + `TypedMergeFilter` / `UniformMergeFilter` subclass), or with the new + overloads `registerTypedMergeFilter(name, slotNames, merger)` + and `registerMergeFilter(name, slotNames, combiner)`. Unknown or missing + names, a group that names only some slots, and names on a merge that declares + none are build-time diagnostics; type mismatches name the slot + (`slot 'raw' of 'Merge' expects ...`). Positional groups keep working and are + matched by position. `dsl::StageNode::slotNames` records the names, and + `dsl::toMermaid` labels the links with them. - **`TODO.md`** — the planned work, with what the library does today, the gap, an API sketch and the workaround for each item; the README roadmap summarizes it. @@ -74,6 +86,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the `std::shared_ptr` for composites that forward it. ### Changed +- `MergeStage::mergeInputTypes()` is no longer pure; it defaults to no slot + types, so a merge can declare only names. `MergeFilter` now implements + `MergeStage`, and `AnyMessageFilter` has a new `mergeInputNames()`, which + defaults to none. - **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 diff --git a/EXAMPLE.md b/EXAMPLE.md index 2b3b3cd..2495988 100644 --- a/EXAMPLE.md +++ b/EXAMPLE.md @@ -271,6 +271,49 @@ compiler; they are shortened here.) lambda, and `UniformMergeFilter` covers the other shape: any number of slots, all of the same type (N variants of one computation, combined). +### Named slots + +Slot types cannot catch everything: when two slots have the *same* type, +swapping them still passes validation. A merge can name its slots so that the +group says which edge goes where: + +```cpp +class CompareFilter : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&& before, + std::optional&& after) override + { + return std::format("{} => {}", before.value_or("-"), after.value_or("-")); + } + + std::vector mergeInputNames() const override + { + return {"before", "after"}; + } +}; +static FilterRegistrar registerCompare("Compare"); +``` + +```cpp +DslFilterGraph pipeline(R"dsl( + in -> Uppercase -> upper + upper -> Reverse -> reversed + (after: reversed, before: upper) -> Compare -> compared -> Print(prefix="[named] ") -> out +)dsl"); + +pipeline.filter(std::string{"Hello, filterGraph!"}); +``` + +```text +[named] HELLO, FILTERGRAPH! => !HPARGRETLIF ,OLLEH +``` + +The group lists `after` first, but the slots are matched by name, so `merge()` +still receives `before` first. A misspelled name, or a slot the group leaves +out, is reported when the graph is built. A group without names still feeds +`Compare` by position. + ## 5. Several named outputs A graph can have more than one output. `out.` names each one, and the @@ -779,6 +822,7 @@ The complete output of `textPipeline` (sections 1–7): [typed] HELLO, FILTERGRAPH! (19 chars) [typed check] 3:20: slot 1 of 'Report' expects 'class std::basic_string' but edge 'length' carries 'unsigned __int64' [typed check] 3:20: slot 2 of 'Report' expects 'unsigned __int64' but edge 'upper' carries 'class std::basic_string' +[named] HELLO, FILTERGRAPH! => !HPARGRETLIF ,OLLEH [outputs] upper=HELLO, FILTERGRAPH! length=19 long=dropped [check] 1:7: unknown stage type 'Uppercas' — did you mean 'Uppercase'? [check] 2:7: could not construct 'MinLength': [json.exception.out_of_range.403] key 'minLength' not found diff --git a/README.md b/README.md index 2535172..2b17a22 100644 --- a/README.md +++ b/README.md @@ -254,7 +254,9 @@ edge -> Stage -> edge -> Stage(key=value) -> edge stage, one slot per edge, in the order listed. A merge either declares its slot types (see [Typed merges](#typed-merges)) or takes them untyped as `MergeInputs` (`std::vector`), as - `registerMergeFilter(name, combiner)` does. + `registerMergeFilter(name, combiner)` does. A group may name the + slots of a merge that declares names, `(raw: msg, checked: valid) -> Merge`, + and is then matched by name, in any order (see [Named slots](#named-slots)). - `#` starts a comment that runs to the end of the line. ### How a graph runs @@ -313,6 +315,42 @@ Mis-wiring a group is then a build-time diagnostic, not a wrong result: `MergeFilter` / `registerMergeFilter` declare no slot types; their edges stay unchecked, and the combiner reads the slots with `std::any_cast`. +### Named slots + +Slot types cannot tell two slots of the *same* type apart: written in the wrong +order, they pass every check and run silently wrong. A merge can therefore name +its slots, and a group can refer to them by name: + +```cpp +class Compare : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&& before, std::optional&& after) override { ... } + + std::vector mergeInputNames() const override { return {"before", "after"}; } +}; + +// The same for merges registered from a function: +registerTypedMergeFilter("Compare", {"before", "after"}, merger); +registerMergeFilter("CompareAny", {"before", "after"}, combiner); +``` + +```text +(after: current, before: baseline) -> Compare -> verdict +``` + +- A **named group** is matched by name, so its order does not matter: the merge + receives its slots in the order it declares them. An unknown name + (`'Compare' has no slot named 'befor' — did you mean 'before'?`) and a slot + the group does not feed are build-time diagnostics. +- A **positional group** still works with a merge that names its slots, matched + by position, so existing graphs stay valid. A merge that names its slots has + exactly that many. +- A group may name all of its slots or none. Naming the slots of a merge that + declares no names is an error. + +`dsl::toMermaid` labels the links of a named group with the slot names. + ### Choosing the output type `DslFilterGraph`'s `OutputType` states what the graph's @@ -586,7 +624,6 @@ Planned improvements, not yet implemented, live in [TODO.md](TODO.md) — each with what the library does today, the gap, an API sketch and the workaround available in the meantime. The current list: -- **Named merge slots**, so a fan-in group matches by name, not position. - **Several named graph inputs** (`in.orders`, `in.quotes`), mirroring `GraphOutputs`. - **Stage labels and typed access** (`Summarize@stats`, `graph.stage("stats")`). diff --git a/TODO.md b/TODO.md index e149843..be316b1 100644 --- a/TODO.md +++ b/TODO.md @@ -18,67 +18,26 @@ Examples use the generic stages of the [README](README.md) (`Parse`, | # | Item | Impact | Compatibility | Workaround today | |---|------|--------|---------------|------------------| -| 1 | [Named merge slots](#1-named-merge-slots) | medium | additive if done carefully | rely on group order | -| 2 | [Several named graph inputs](#2-several-named-graph-inputs) | medium | additive (new graph shape) | `std::variant` input + select stages | -| 3 | [Stage labels and typed access](#3-stage-labels-and-typed-access) | medium | additive (DSL syntax) | side registry filled by creator lambdas | -| 4 | [Track which stage short-circuited](#4-track-which-stage-short-circuited) | medium | additive | tap the edge, or log in the stage | -| 5 | [Config-aware `registerMergeFilter`](#5-config-aware-registermergefilter) | low | additive | subclass + `FilterRegistrar` creator | -| 6 | [Injectable registry, duplicate detection](#6-injectable-registry-duplicate-detection) | low | mostly additive | unique names | -| 7 | [Documentation: 0..n outputs, large messages](#7-documentation-0n-outputs-large-messages) | doc only | — | — | -| 8 | [Known limitations to lift](#8-known-limitations-to-lift) | low–medium | additive | JSON config; read `GraphOutputs` carefully | +| 1 | [Several named graph inputs](#1-several-named-graph-inputs) | medium | additive (new graph shape) | `std::variant` input + select stages | +| 2 | [Stage labels and typed access](#2-stage-labels-and-typed-access) | medium | additive (DSL syntax) | side registry filled by creator lambdas | +| 3 | [Track which stage short-circuited](#3-track-which-stage-short-circuited) | medium | additive | tap the edge, or log in the stage | +| 4 | [Config-aware `registerMergeFilter`](#4-config-aware-registermergefilter) | low | additive | subclass + `FilterRegistrar` creator | +| 5 | [Injectable registry, duplicate detection](#5-injectable-registry-duplicate-detection) | low | mostly additive | unique names | +| 6 | [Documentation: 0..n outputs, large messages](#6-documentation-0n-outputs-large-messages) | doc only | — | — | +| 7 | [Known limitations to lift](#7-known-limitations-to-lift) | low–medium | additive | JSON config; read `GraphOutputs` carefully | Items of the same list that are done, and therefore not repeated here: **type-checked merge inputs** (`TypedMergeFilter` / `UniformMergeFilter` / -`registerTypedMergeFilter`), the **end-of-stream hook** -(`MessageFilter::finish()`), both described in the README, and **examples for +`registerTypedMergeFilter`), **named merge slots** +(`(raw: msg, checked: valid) -> Merge`, `MergeStage::mergeInputNames()`), the +**end-of-stream hook** (`MessageFilter::finish()`), all described in the +README, and **examples for the newer features** — `apps/statefulPipeline` and `apps/compositePipeline`, walked through in [EXAMPLE.md](EXAMPLE.md) sections 8 and 9. --- -## 1. Named merge slots - -**Today.** Slots are positional, in the order the DSL group lists them. A merge -stage can only know which upstream a slot came from by convention. Since -type-checked merges shipped, a mis-*typed* slot is caught when the graph is -built — but two slots of the *same* type in the wrong order still pass -validation and then run silently wrong. - -**Proposal.** Optional slot names in the group, checked against the names the -stage declares: - -```text -(raw: msg, checked: valid) -> Summarize -> out.stats -``` - -```cpp -// empty = positional, today's behaviour -virtual std::vector mergeInputNames() const { return {}; } -``` - -Behaviour: - -- **Stage declares names, group uses names:** slots are matched by name, so the - order in the group no longer matters. Unknown or missing names produce - diagnostics. -- **Stage declares names, group is positional:** allowed, matched by position - (keeps today's graphs valid). -- **Stage declares nothing:** names in the group are an error (*'Summarize' has - no named slots*). - -Together with the slot types a typed merge already declares, this gives every -slot a name *and* a type, which also makes `dsl::toMermaid` output -self-explanatory. - -**Compatibility.** Keep `MergeInputs` a `std::vector` and reorder the -values into declared order before calling the stage, so stage code does not -change. Making `MergeInputs` a struct with names would be a breaking change and -is not needed. - -**Workaround today.** Rely on group order, and give same-typed slots distinct -wrapper types so that the existing slot-type check can tell them apart. - -## 2. Several named graph inputs +## 1. Several named graph inputs **Today.** A graph has exactly one input edge `in`, with one type. A graph that consumes several kinds of message needs a single `std::variant` input, and then @@ -113,7 +72,7 @@ keeps the single `in` edge. **Workaround today.** A `std::variant` input plus one select stage per alternative, each dropping the alternatives it does not handle. -## 3. Stage labels and typed access +## 2. Stage labels and typed access **Today.** The owner of a `DslFilterGraph` cannot reach a stage instance. The DSL has no labels (`#` starts a comment), and the compiled plan is private. @@ -143,7 +102,7 @@ auto& stats = graph.stage("stats"); // throws if unknown or the every instance it builds in a side registry, or have the stage publish what the owner needs through the `GraphContext`. -## 4. Track which stage short-circuited +## 3. Track which stage short-circuited **Today.** When a graph drops a message, neither `DslFilterGraph` nor `AnyFilterChain` tells the caller *which* stage returned `std::nullopt`. @@ -164,7 +123,7 @@ so that variant needs an overload or an opt-in. **Workaround today.** Tap the suspect edge with a logging stage ending in `end`, or log inside the stage that decides to drop. -## 5. Config-aware `registerMergeFilter` +## 4. Config-aware `registerMergeFilter` **Today.** `registerMergeFilter(name, combiner)` ignores the DSL config and **copies one combiner into every instance**. A combiner lambda that captures a @@ -194,7 +153,7 @@ one. **Workaround today.** Subclass and register with a creator lambda. -## 6. Injectable registry, duplicate detection +## 5. Injectable registry, duplicate detection **Today.** `FilterRegistry::instance()` is a process-wide singleton, and `registerFilter` **silently overwrites** an existing name. @@ -223,7 +182,7 @@ Ship it behind a transition: warn first, or add **Workaround today.** Keep names unique, and register test fakes under their own names. -## 7. Documentation: 0..n outputs, large messages +## 6. Documentation: 0..n outputs, large messages No API change. These points belong in the README / EXAMPLE.md, because they come up as soon as stages carry state: @@ -240,7 +199,7 @@ up as soon as stages carry state: its members persists across messages. Say explicitly that this is supported and intended, and that each graph construction creates fresh instances. -## 8. Known limitations to lift +## 7. Known limitations to lift The [current limitations](README.md#current-limitations) the README lists, as work items: diff --git a/apps/textPipeline/main.cpp b/apps/textPipeline/main.cpp index 2c908ed..6bcb0f3 100644 --- a/apps/textPipeline/main.cpp +++ b/apps/textPipeline/main.cpp @@ -3,7 +3,8 @@ // - MessageFilter: the base stage interface // - FilterGraph: compile-time chaining of stages // - FilterRegistry + DslFilterGraph: runtime graphs described in the text DSL, -// with fan-out (a "tap"), fan-in (a merge) and several named outputs +// with fan-out (a "tap"), fan-in (merges, with typed and named slots) and +// several named outputs // - validateDslGraph: every problem in a broken graph, located by line:column // - JsonFilterGraph: the JSON format, which remains supported #include @@ -24,6 +25,7 @@ #include #include #include +#include using filterGraph::DslFilterGraph; using filterGraph::FilterGraph; @@ -173,6 +175,26 @@ class ReportFilter : public TypedMergeFilter registerReport("Report"); +// A merge with two slots of the same type, where the order matters. The slot +// types cannot tell them apart, so it names them: a group can then match them +// by name, `(after: x, before: y) -> Compare`, in any order. +class CompareFilter : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&& before, + std::optional&& after) override + { + return std::format("{} => {}", before.value_or("-"), after.value_or("-")); + } + + std::vector mergeInputNames() const override + { + return {"before", "after"}; + } +}; + +static FilterRegistrar registerCompare("Compare"); + // Only needed by the JSON example: in the DSL, fan-out is built in. static const bool sRegisterFanout = [] { registerFanoutFilter("Fanout"); @@ -240,6 +262,19 @@ int main() } } + // 3c) Named slots: Compare's two slots have the same type, so only their + // names say which is which. The group lists them in the other order; + // the merge still receives `before` first. + { + DslFilterGraph pipeline(R"dsl( + in -> Uppercase -> upper + upper -> Reverse -> reversed + (after: reversed, before: upper) -> Compare -> compared -> Print(prefix="[named] ") -> out + )dsl"); + + pipeline.filter(std::string{"Hello, filterGraph!"}); + } + // 4) Several named outputs of different types, returned as GraphOutputs. { DslFilterGraph analysis(R"dsl( diff --git a/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp b/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp index 4be3cd4..d03a47c 100644 --- a/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp +++ b/libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp @@ -6,8 +6,10 @@ #include #include #include +#include #include #include +#include namespace filterGraph { @@ -42,6 +44,13 @@ class AnyMessageFilter { return {}; } + + // The slot names of a merge stage (see MergeStage); empty for a merge with + // positional slots and for every stage that is not a merge. + virtual std::vector mergeInputNames() const + { + return {}; + } }; // Adapts a concrete MessageFilter to the @@ -95,6 +104,15 @@ class AnyMessageFilterAdapter : public AnyMessageFilter return {}; } + std::vector mergeInputNames() const override + { + if (const auto* merge = dynamic_cast(mFilter.get())) + { + return merge->mergeInputNames(); + } + return {}; + } + private: std::shared_ptr mFilter; }; diff --git a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp index beda7e8..b05bc89 100644 --- a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp @@ -39,7 +39,8 @@ // empty edge are skipped, so the drop propagates downstream; // - a merge `(a, b) -> Merge` receives one MergeInputs slot per edge, with // empty slots ("holes") for dropped edges; it is skipped only when every -// edge is empty; +// edge is empty. The slots are in group order, or, for a group that names +// them, `(raw: a, checked: b)`, in the order the merge declares its names; // - values routed to `end` are discarded; a stage producing Void must route // to `end`. namespace filterGraph { @@ -229,51 +230,164 @@ inline std::string describeEdge(const std::string& edge) return std::format("edge '{}'", edge); } -// Checks the edges of a fan-in group against the slot types the merge stage -// declares (MergeStage); an untyped merge declares none and is not checked. -inline void checkMergeInputs(const StageNode& node, - const MergeSlotTypes& slots, - const std::unordered_map& edgeTypes, - std::vector& diagnostics) +inline std::string joinNames(const std::vector& names) +{ + std::string text; + for (const auto& name : names) + { + text += text.empty() ? name : ", " + name; + } + return text; +} + +// Checks the slot names a merge stage declares (MergeStage) against each other +// and against its slot types. +inline bool checkDeclaredSlotNames(const StageNode& node, + const MergeSlotTypes& types, + const std::vector& names, + std::vector& diagnostics) +{ + if (!types.empty() && !types.uniform && types.types.size() != names.size()) + { + diagnostics.push_back({node.loc, + std::format("stage '{}' declares {} slot names for {} slot types", + node.type, + names.size(), + types.types.size())}); + return false; + } + std::unordered_set seen; + for (const auto& name : names) + { + if (!seen.insert(name).second) + { + diagnostics.push_back({node.loc, std::format("stage '{}' declares slot '{}' twice", node.type, name)}); + return false; + } + } + return true; +} + +// Matches the named slots of a fan-in group to the names the merge stage +// declares, and returns the group's edges in the stage's slot order. +inline std::optional> matchSlotNames(const StageNode& node, + const std::vector& names, + std::vector& diagnostics) +{ + if (names.empty()) + { + diagnostics.push_back( + {node.loc, + std::format("'{}' has no named slots; list the group's edges without names", node.type)}); + return std::nullopt; + } + + bool matched = true; + for (const auto& slot : node.slotNames) + { + if (std::find(names.begin(), names.end(), slot) == names.end()) + { + std::string message = std::format("'{}' has no slot named '{}'", node.type, slot); + if (auto suggestion = filterGraph::detail::closestName(slot, names)) + { + message += std::format(" \u2014 did you mean '{}'?", *suggestion); + } + message += std::format(" (its slots: {})", joinNames(names)); + diagnostics.push_back({node.loc, std::move(message)}); + matched = false; + } + } + for (const auto& name : names) + { + if (std::find(node.slotNames.begin(), node.slotNames.end(), name) == node.slotNames.end()) + { + diagnostics.push_back( + {node.loc, std::format("the fan-in group does not feed slot '{}' of '{}'", name, node.type)}); + matched = false; + } + } + if (!matched) + { + return std::nullopt; + } + + // Both name lists are unique and contain each other: a permutation. + std::vector edges; + edges.reserve(names.size()); + for (const auto& name : names) + { + const auto position = std::find(node.slotNames.begin(), node.slotNames.end(), name) - node.slotNames.begin(); + edges.push_back(node.inputs[static_cast(position)]); + } + return edges; +} + +// Wires a fan-in group to the slots its merge stage declares (MergeStage): +// matches named slots by name and positional ones by position, checks the +// edges against the declared slot types, and returns the group's edges in the +// stage's slot order. An untyped merge with positional slots is not checked. +// Problems are appended to `diagnostics`; the plan must then not be run. +inline std::vector wireMergeSlots(const StageNode& node, + const MergeSlotTypes& types, + const std::vector& names, + const std::unordered_map& edgeTypes, + std::vector& diagnostics) { - if (slots.empty()) + if (!names.empty() && !checkDeclaredSlotNames(node, types, names, diagnostics)) { - return; + return node.inputs; + } + + std::vector edges = node.inputs; + if (!node.slotNames.empty()) + { + auto ordered = matchSlotNames(node, names, diagnostics); + if (!ordered) + { + return node.inputs; + } + edges = std::move(*ordered); } - if (!slots.uniform && slots.types.size() != node.inputs.size()) + // Named slots fix the count, even of an untyped or uniform merge. + const std::size_t slots = !names.empty() ? names.size() + : !types.empty() && !types.uniform ? types.types.size() + : edges.size(); + if (slots != edges.size()) { // Without a slot-to-edge correspondence, per-slot messages would only // repeat this one. diagnostics.push_back({node.loc, std::format("stage '{}' takes {} inputs but the group has {}", node.type, - slots.types.size(), - node.inputs.size())}); - return; + slots, + edges.size())}); + return node.inputs; } - for (std::size_t slot = 0; slot < node.inputs.size(); ++slot) + for (std::size_t slot = 0; slot < edges.size() && !types.empty(); ++slot) { - const std::type_index expected = slots.uniform ? slots.types.front() : slots.types[slot]; - const auto type = edgeTypes.find(node.inputs[slot]); + const std::type_index expected = types.uniform ? types.types.front() : types.types[slot]; + const auto type = edgeTypes.find(edges[slot]); if (type != edgeTypes.end() && type->second != expected) { + const std::string label = names.empty() ? std::to_string(slot + 1) : std::format("'{}'", names[slot]); diagnostics.push_back({node.loc, std::format("slot {} of '{}' expects '{}' but {} carries '{}'", - slot + 1, + label, node.type, expected.name(), - describeEdge(node.inputs[slot]), + describeEdge(edges[slot]), type->second.name())}); } } + return edges; } struct CompiledStage { std::shared_ptr filter; - std::vector inputs; // edge slots read by the stage + std::vector inputs; // edge slots read by the stage (a merge's in slot order) std::optional output; // edge slot written; nullopt => `end` bool merge = false; // inputs are gathered into MergeInputs }; @@ -333,7 +447,8 @@ class GraphPlan continue; } - const bool merge = filter->inputType() == std::type_index(typeid(MergeInputs)); + const bool merge = filter->inputType() == std::type_index(typeid(MergeInputs)); + std::vector inputs = node.inputs; if (node.fanIn && !merge) { diagnostics.push_back( @@ -352,7 +467,8 @@ class GraphPlan } else if (merge) { - checkMergeInputs(node, filter->mergeInputTypes(), edgeTypes, diagnostics); + inputs = wireMergeSlots(node, filter->mergeInputTypes(), filter->mergeInputNames(), edgeTypes, + diagnostics); } else { @@ -386,7 +502,7 @@ class GraphPlan CompiledStage stage; stage.filter = std::move(filter); stage.merge = merge; - for (const auto& edge : node.inputs) + for (const auto& edge : inputs) { stage.inputs.push_back(slotOf(edge)); } diff --git a/libs/filterGraph/core/filterGraph/GraphLang.hpp b/libs/filterGraph/core/filterGraph/GraphLang.hpp index 4bc1da6..263af1f 100644 --- a/libs/filterGraph/core/filterGraph/GraphLang.hpp +++ b/libs/filterGraph/core/filterGraph/GraphLang.hpp @@ -43,7 +43,9 @@ // optionally with config args `Name(k=v, k2="s", k3=true)`; // - fan-out = reuse an edge name as a source in several statements; // - fan-in = a source group `(a, b) -> Merge -> c`, where Merge is a merge -// stage (see registerMergeFilter in MergeFilter.hpp); +// stage (see registerMergeFilter in MergeFilter.hpp); the group may name the +// merge's slots, `(raw: a, checked: b) -> Merge -> c`, which matches them by +// name instead of by position; // - `#` starts a comment that runs to the end of the line. // // This header parses (with a lexy-based parser) and structurally validates a @@ -73,6 +75,7 @@ struct StageNode std::vector inputs; // source edge names ("in" allowed) std::optional output; // produced edge name ("$outN" for `out`); nullopt => routed to end bool fanIn = false; // inputs came from a group `(a, b) -> Stage` + std::vector slotNames; // per input, from `(name: edge, ...)`; empty for a positional group SourceLoc loc; }; @@ -282,7 +285,7 @@ inline Found describeFound(std::string_view rest) } return {false, std::format("'{}'", rest.substr(0, length))}; } - if (c == '(' || c == ')' || c == ',' || c == '=' || c == '.') + if (c == '(' || c == ')' || c == ',' || c == '=' || c == '.' || c == ':') { return {false, std::format("'{}'", c)}; } @@ -307,9 +310,37 @@ struct Term std::optional key; // for `out.` nlohmann::json config; // for Stage std::vector edges; // for Group + std::vector slotNames; // for Group: per edge, from `name: edge`; "" when unnamed SourceLoc loc; }; +// Checks the slot names of a fan-in group: all slots are named or none, and no +// name is used twice. +inline bool checkSlotNames(const Term& group, std::vector& diags) +{ + const auto named = std::count_if(group.slotNames.begin(), group.slotNames.end(), + [](const std::string& name) { return !name.empty(); }); + if (named == 0) + { + return true; + } + if (static_cast(named) != group.slotNames.size()) + { + diags.push_back({group.loc, "a fan-in group names either all of its slots or none, e.g. '(a: x, b: y)'"}); + return false; + } + std::unordered_set seen; + for (const auto& name : group.slotNames) + { + if (!seen.insert(name).second) + { + diags.push_back({group.loc, std::format("slot '{}' is named twice in the fan-in group", name)}); + return false; + } + } + return true; +} + // Interprets a flat term list (edge, stage, edge, ...) into stage nodes, wiring // each stage's inputs from the previous edge/group and its output edge from the // next edge/boundary. Shared by every parser. @@ -371,6 +402,10 @@ inline void buildStatement(std::vector& terms, diags.push_back({terms[i].loc, "a fan-in group needs at least one edge"}); return; } + if (terms[i].kind == Term::Kind::Group && !checkSlotNames(terms[i], diags)) + { + return; + } } } @@ -389,6 +424,10 @@ inline void buildStatement(std::vector& terms, { node.inputs = source.edges; node.fanIn = true; + if (!source.slotNames.front().empty()) + { + node.slotNames = source.slotNames; + } } else { @@ -800,7 +839,21 @@ inline bool parseStatementLine(std::string_view lineText, { return unexpected("expected an edge name in group"); } - term.edges.push_back(captureIdent()); + // `edge`, or `name: edge` for a named slot. + std::string edge = captureIdent(); + std::string slot; + skipBlank(); + if (sc.branch(ld::lit_c<':'>)) + { + skipBlank(); + if (!sc.peek(ld::ascii::alpha_underscore)) + { + return unexpected(std::format("expected an edge name after '{}:'", edge)); + } + slot = std::exchange(edge, captureIdent()); + } + term.edges.push_back(std::move(edge)); + term.slotNames.push_back(std::move(slot)); } }; @@ -1000,9 +1053,18 @@ inline std::string toMermaid(const GraphProgram& program) } nodes += std::format(" s{}[\"{}\"]\n", i, label); - for (const auto& input : stage.inputs) + for (std::size_t slot = 0; slot < stage.inputs.size(); ++slot) { - links += std::format(" {} --> s{}\n", edgeId(input), i); + const std::string& input = stage.inputs[slot]; + if (stage.slotNames.empty()) + { + links += std::format(" {} --> s{}\n", edgeId(input), i); + } + else + { + links += std::format(" {} -->|\"{}\"| s{}\n", edgeId(input), + detail::mermaidLabel(stage.slotNames[slot]), i); + } } if (stage.output) { diff --git a/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp b/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp index b10d751..21b15f6 100644 --- a/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp +++ b/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp @@ -10,6 +10,7 @@ #include #include #include +#include #include // DEPRECATED: the original hand-written tokenizer and recursive-descent parser @@ -32,6 +33,7 @@ enum class TokenKind Comma, // , Equals, // = Dot, // . + Colon, // : Invalid, // an unexpected character or an unterminated string Newline, EndOfInput @@ -83,13 +85,14 @@ class Lexer tokens.push_back(take(TokenKind::Arrow, 2)); continue; } - if (c == '(' || c == ')' || c == ',' || c == '=' || c == '.') + if (c == '(' || c == ')' || c == ',' || c == '=' || c == '.' || c == ':') { const TokenKind kind = c == '(' ? TokenKind::LParen : c == ')' ? TokenKind::RParen : c == ',' ? TokenKind::Comma : c == '=' ? TokenKind::Equals - : TokenKind::Dot; + : c == '.' ? TokenKind::Dot + : TokenKind::Colon; tokens.push_back(take(kind, 1)); continue; } @@ -321,8 +324,21 @@ class Parser unexpected("expected an edge name in group"); return term; } - term.edges.push_back(peek().text); - next(); + // `edge`, or `name: edge` for a named slot. + std::string edge = next().text; + std::string slot; + if (peek().kind == TokenKind::Colon) + { + next(); + if (peek().kind != TokenKind::Identifier) + { + unexpected(std::format("expected an edge name after '{}:'", edge)); + return term; + } + slot = std::exchange(edge, next().text); + } + term.edges.push_back(std::move(edge)); + term.slotNames.push_back(std::move(slot)); } } diff --git a/libs/filterGraph/core/filterGraph/MergeFilter.hpp b/libs/filterGraph/core/filterGraph/MergeFilter.hpp index 22a0e42..71a7240 100644 --- a/libs/filterGraph/core/filterGraph/MergeFilter.hpp +++ b/libs/filterGraph/core/filterGraph/MergeFilter.hpp @@ -31,15 +31,19 @@ namespace filterGraph { // group's edges into it. // // The combiner has the same signature as JoinFilter's, so a combiner written -// for a JSON Join can be reused as a DSL merge unchanged. +// for a JSON Join can be reused as a DSL merge unchanged. Its slots are untyped; +// they are named if `slotNames` is given (see MergeStage::mergeInputNames). template -class MergeFilter : public MessageFilter +class MergeFilter + : public MessageFilter + , public MergeStage { public: using Combiner = std::function(MergeInputs&&)>; - explicit MergeFilter(Combiner combiner) + explicit MergeFilter(Combiner combiner, std::vector slotNames = {}) : mCombiner(std::move(combiner)) + , mSlotNames(std::move(slotNames)) { } @@ -48,8 +52,14 @@ class MergeFilter : public MessageFilter return mCombiner(std::move(inputs)); } + std::vector mergeInputNames() const override + { + return mSlotNames; + } + private: - Combiner mCombiner; + Combiner mCombiner; + std::vector mSlotNames; }; // Registers MergeFilter under `name` for use as a DSL fan-in stage: @@ -69,13 +79,34 @@ void registerMergeFilter(const std::string& name, typename MergeFilter("Summarize", {"raw", "checked"}, combiner); +// (checked: valid, raw: msg) -> Summarize -> stats +// +// The combiner receives the slots in the order of `slotNames`, whatever the +// order of the group. +template +void registerMergeFilter(const std::string& name, + std::vector slotNames, + typename MergeFilter::Combiner combiner) +{ + FilterRegistrar> registrar( + name, + [combiner = std::move(combiner), slotNames = std::move(slotNames)](const nlohmann::json&) { + return std::make_shared>(combiner, slotNames); + }); +} + // TypedMergeFilter is a merge stage that declares // the type of every one of its slots, so that the DSL checks the edges of its // fan-in group when the graph is built instead of failing with a // std::bad_any_cast on the first message. // -// The slots arrive as std::optionals, in group order: an empty optional is a -// hole left by a path that dropped the message. Implement merge(): +// The slots arrive as std::optionals, in slot order: an empty optional is a +// hole left by a path that dropped the message. Override mergeInputNames() to +// name the slots, so that a group can match them by name, in any order (see +// MergeStage). Implement merge(): // // class Summarize : public TypedMergeFilter // { @@ -169,8 +200,9 @@ class CallableTypedMergeFilter : public TypedMergeFilter(std::optional&&...)>; - explicit CallableTypedMergeFilter(Merger merger) + explicit CallableTypedMergeFilter(Merger merger, std::vector slotNames = {}) : mMerger(std::move(merger)) + , mSlotNames(std::move(slotNames)) { } @@ -179,8 +211,14 @@ class CallableTypedMergeFilter : public TypedMergeFilter mergeInputNames() const override + { + return mSlotNames; + } + private: - Merger mMerger; + Merger mMerger; + std::vector mSlotNames; }; } // namespace detail @@ -204,4 +242,20 @@ void registerTypedMergeFilter(const std::string& }); } +// Registers a typed merge with named slots, one name per slot type: +// +// registerTypedMergeFilter("Summarize", {"message", "valid"}, merger); +// (valid: checked, message: msg) -> Summarize -> stats +template +void registerTypedMergeFilter(const std::string& name, + std::vector slotNames, + typename detail::CallableTypedMergeFilter::Merger merger) +{ + using Stage = detail::CallableTypedMergeFilter; + FilterRegistrar registrar( + name, [merger = std::move(merger), slotNames = std::move(slotNames)](const nlohmann::json&) { + return std::make_shared(merger, slotNames); + }); +} + } // namespace filterGraph diff --git a/libs/filterGraph/core/filterGraph/MergeStage.hpp b/libs/filterGraph/core/filterGraph/MergeStage.hpp index eaac124..5f085e1 100644 --- a/libs/filterGraph/core/filterGraph/MergeStage.hpp +++ b/libs/filterGraph/core/filterGraph/MergeStage.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -29,14 +30,28 @@ struct MergeSlotTypes }; // Implemented by merge stages (MessageFilter) that declare -// their slot types. TypedMergeFilter and UniformMergeFilter implement it; a -// hand-written merge stage may derive from it too. +// their slots: their types, their names, or both. TypedMergeFilter, +// UniformMergeFilter and MergeFilter implement it; a hand-written merge stage +// may derive from it too. class MergeStage { public: virtual ~MergeStage() = default; - virtual MergeSlotTypes mergeInputTypes() const = 0; + // The slot types; empty (the default) for an untyped merge. + virtual MergeSlotTypes mergeInputTypes() const + { + return {}; + } + + // The slot names, in slot order; empty (the default) for a merge whose + // slots are positional. A merge that names its slots has exactly that many, + // and a DSL group may then match them by name, `(raw: msg, checked: valid)`, + // in any order: the slots are handed over in the order declared here. + virtual std::vector mergeInputNames() const + { + return {}; + } }; } // namespace filterGraph diff --git a/tests/FilterGraphTests/FilterGraphTests.cpp b/tests/FilterGraphTests/FilterGraphTests.cpp index 7c26489..67f9cdc 100644 --- a/tests/FilterGraphTests/FilterGraphTests.cpp +++ b/tests/FilterGraphTests/FilterGraphTests.cpp @@ -933,6 +933,11 @@ std::string describeProgram(const dsl::GraphProgram& program) { text += input + ','; } + text += " slots="; + for (const auto& slot : stage.slotNames) + { + text += slot + ','; + } text += std::format(" output={} fanIn={}\n", stage.output.value_or(""), stage.fanIn); } for (const auto& output : program.outputs) @@ -992,6 +997,11 @@ TEST_CASE("Both parsers report the same located syntax errors", "[GraphDslErrors {"in -> RunDouble -> -", "1:20: unexpected character '-'"}, {"in -> RunDouble -> \xC3\xA9", "1:20: unexpected byte 0xC3"}, {"in -> RunDouble RunDouble -> out\r\n", "1:17: expected '->' but found 'RunDouble'"}, + {"(a: -> RunDouble -> out", "1:5: expected an edge name after 'a:' but found '->'"}, + {"(a: 3) -> RunDouble -> out", "1:5: expected an edge name after 'a:' but found '3'"}, + {"(a b: c) -> RunDouble -> out", "1:1: a fan-in group names either all of its slots or none, e.g. '(a: x, b: y)'"}, + {"(a: x, a: y) -> RunDouble -> out", "1:1: slot 'a' is named twice in the fan-in group"}, + {"in -> RunDouble: -> out", "1:16: expected '->' but found ':'"}, }; for (const auto& [source, expected] : cases) diff --git a/tests/FilterGraphTests/NamedSlotTests.cpp b/tests/FilterGraphTests/NamedSlotTests.cpp new file mode 100644 index 0000000..e067feb --- /dev/null +++ b/tests/FilterGraphTests/NamedSlotTests.cpp @@ -0,0 +1,299 @@ +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace filterGraph; + +namespace { + +class PlusOneFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return value + 1; + } +}; + +class TimesTenFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return value * 10; + } +}; + +class ToStringFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return std::to_string(value); + } +}; + +class DropOddFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return value % 2 == 0 ? std::optional(value) : std::nullopt; + } +}; + +// Two slots of the same type, where the order matters: "-", with +// "?" for a hole. +class DifferenceMerge : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&& base, std::optional&& bonus) override + { + return std::format("{}-{}", base ? std::to_string(*base) : "?", bonus ? std::to_string(*bonus) : "?"); + } + + std::vector mergeInputNames() const override + { + return {"base", "bonus"}; + } +}; + +// Two slots of different types, named. +class LabelMerge : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&& text, std::optional&& number) override + { + return std::format("{}/{}", text.value_or("-"), number.value_or(0)); + } + + std::vector mergeInputNames() const override + { + return {"text", "number"}; + } +}; + +// Declares more names than it has slot types: a bug in the stage. +class BrokenNamesMerge : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&&, std::optional&&) override + { + return 0; + } + + std::vector mergeInputNames() const override + { + return {"a", "b", "c"}; + } +}; + +// Untyped and unnamed: positional slots only. +class PositionalMerge : public UniformMergeFilter +{ +public: + std::optional merge(std::vector>&& inputs) override + { + return static_cast(inputs.size()); + } +}; + +static FilterRegistrar registerPlusOne("SlotPlusOne"); +static FilterRegistrar registerTimesTen("SlotTimesTen"); +static FilterRegistrar registerToString("SlotToString"); +static FilterRegistrar registerDropOdd("SlotDropOdd"); +static FilterRegistrar registerDifference("SlotDifference"); +static FilterRegistrar registerLabel("SlotLabel"); +static FilterRegistrar registerBroken("SlotBroken"); +static FilterRegistrar registerPositional("SlotPositional"); + +static const bool sRegisterCallables = [] { + registerMergeFilter("SlotUntyped", {"first", "second"}, [](MergeInputs&& inputs) { + return std::optional(std::format("{},{}", std::any_cast(inputs[0]), std::any_cast(inputs[1]))); + }); + registerTypedMergeFilter( + "SlotLambda", + {"number", "text"}, + [](std::optional&& number, std::optional&& text) -> std::optional { + return std::format("{}:{}", number.value_or(0), text.value_or("-")); + }); + return true; +}(); + +std::vector buildErrors(std::string_view text) +{ + return validateDslGraph(text); +} + +bool mentions(const std::vector& diagnostics, std::string_view fragment) +{ + return std::any_of(diagnostics.begin(), diagnostics.end(), [&](const dsl::TextDiagnostic& diagnostic) { + return diagnostic.message.find(fragment) != std::string::npos; + }); +} + +} // namespace + +TEST_CASE("A named group matches slots by name, in any order", "[NamedSlots]") +{ + const char* inOrder = R"( + in -> SlotPlusOne -> small + in -> SlotTimesTen -> large + (base: large, bonus: small) -> SlotDifference -> out + )"; + const char* swapped = R"( + in -> SlotPlusOne -> small + in -> SlotTimesTen -> large + (bonus: small, base: large) -> SlotDifference -> out + )"; + + REQUIRE(DslFilterGraph(inOrder).filter(4).value() == "40-5"); + REQUIRE(DslFilterGraph(swapped).filter(4).value() == "40-5"); +} + +TEST_CASE("A positional group still feeds a named merge by position", "[NamedSlots]") +{ + DslFilterGraph graph(R"( + in -> SlotPlusOne -> small + in -> SlotTimesTen -> large + (small, large) -> SlotDifference -> out + )"); + + REQUIRE(graph.filter(4).value() == "5-40"); +} + +TEST_CASE("Named slots keep their holes when reordered", "[NamedSlots]") +{ + DslFilterGraph graph(R"( + in -> SlotDropOdd -> even + in -> SlotTimesTen -> large + (bonus: even, base: large) -> SlotDifference -> out + )"); + + REQUIRE(graph.filter(2).value() == "20-2"); + REQUIRE(graph.filter(3).value() == "30-?"); +} + +TEST_CASE("Named slots work for untyped and lambda-registered merges", "[NamedSlots]") +{ + DslFilterGraph untyped(R"( + in -> SlotPlusOne -> small + in -> SlotTimesTen -> large + (second: small, first: large) -> SlotUntyped -> out + )"); + REQUIRE(untyped.filter(1).value() == "10,2"); + + DslFilterGraph lambda(R"( + in -> SlotToString -> text + in -> SlotTimesTen -> large + (text: text, number: large) -> SlotLambda -> out + )"); + REQUIRE(lambda.filter(3).value() == "30:3"); +} + +TEST_CASE("A slot type mismatch names the slot", "[NamedSlots]") +{ + const auto diagnostics = buildErrors(R"( + in -> SlotToString -> text + in -> SlotTimesTen -> large + (text: large, number: text) -> SlotLabel -> out + )"); + + REQUIRE(diagnostics.size() == 2); + REQUIRE(diagnostics[0].message.starts_with("slot 'text' of 'SlotLabel' expects")); + REQUIRE(diagnostics[0].message.find("edge 'large' carries") != std::string::npos); + REQUIRE(diagnostics[1].message.starts_with("slot 'number' of 'SlotLabel' expects")); + REQUIRE(diagnostics[0].loc.line == 4); +} + +TEST_CASE("Unknown and missing slot names are reported", "[NamedSlots]") +{ + const auto unknown = buildErrors(R"( + in -> SlotPlusOne -> small + in -> SlotTimesTen -> large + (base: large, bonsu: small) -> SlotDifference -> out + )"); + REQUIRE(unknown.size() == 2); + REQUIRE(unknown[0].message.starts_with("'SlotDifference' has no slot named 'bonsu'")); + REQUIRE(unknown[0].message.ends_with("did you mean 'bonus'? (its slots: base, bonus)")); + REQUIRE(unknown[1].message == "the fan-in group does not feed slot 'bonus' of 'SlotDifference'"); + + const auto missing = buildErrors(R"( + in -> SlotTimesTen -> large + (base: large) -> SlotDifference -> out + )"); + REQUIRE(missing.size() == 1); + REQUIRE(missing[0].message == "the fan-in group does not feed slot 'bonus' of 'SlotDifference'"); +} + +TEST_CASE("A named group needs a merge that names its slots", "[NamedSlots]") +{ + const auto diagnostics = buildErrors(R"( + in -> SlotPlusOne -> a + in -> SlotTimesTen -> b + (x: a, y: b) -> SlotPositional -> n -> SlotToString -> out + )"); + + REQUIRE(diagnostics.size() == 1); + REQUIRE(diagnostics[0].message == "'SlotPositional' has no named slots; list the group's edges without names"); +} + +TEST_CASE("A merge that names its slots fixes their number", "[NamedSlots]") +{ + const auto diagnostics = buildErrors(R"( + in -> SlotPlusOne -> a + in -> SlotTimesTen -> b + in -> SlotTimesTen -> c + (a, b, c) -> SlotUntyped -> out + )"); + + REQUIRE(diagnostics.size() == 1); + REQUIRE(diagnostics[0].message == "stage 'SlotUntyped' takes 2 inputs but the group has 3"); +} + +TEST_CASE("Inconsistent slot declarations of a stage are reported", "[NamedSlots]") +{ + const auto diagnostics = buildErrors(R"( + in -> SlotPlusOne -> a + in -> SlotTimesTen -> b + (a, b) -> SlotBroken -> n -> SlotToString -> out + )"); + + REQUIRE(diagnostics.size() == 1); + REQUIRE(diagnostics[0].message == "stage 'SlotBroken' declares 3 slot names for 2 slot types"); +} + +TEST_CASE("The parsed program and toMermaid carry the slot names", "[NamedSlots]") +{ + const auto program = dsl::parseGraphProgram("in -> SlotPlusOne -> small\n" + "in -> SlotTimesTen -> large\n" + "(bonus: small, base: large) -> SlotDifference -> out\n"); + REQUIRE(program.ok()); + REQUIRE(program.stages[2].slotNames == std::vector{"bonus", "base"}); + REQUIRE(program.stages[2].inputs == std::vector{"small", "large"}); + REQUIRE(program.stages[0].slotNames.empty()); + + const std::string mermaid = dsl::toMermaid(program); + REQUIRE(mermaid.find("-->|\"bonus\"| s2") != std::string::npos); + REQUIRE(mermaid.find("-->|\"base\"| s2") != std::string::npos); +} + +TEST_CASE("Merge stages expose their declared slot names", "[NamedSlots]") +{ + REQUIRE(DifferenceMerge{}.mergeInputNames() == std::vector{"base", "bonus"}); + REQUIRE(PositionalMerge{}.mergeInputNames().empty()); + + auto adapted = FilterRegistry::instance().create("SlotUntyped", nlohmann::json::object()); + REQUIRE(adapted->mergeInputNames() == std::vector{"first", "second"}); + REQUIRE(adapted->mergeInputTypes().empty()); +}