From ff42094909a3c30a938275e946a311e63f0c09b9 Mon Sep 17 00:00:00 2001 From: psy_inf Date: Sat, 19 Sep 2026 21:38:47 +0200 Subject: [PATCH] feat: several named graph inputs A graph can read named inputs, `in.`, with the new GraphInputs as its input type, mirroring GraphOutputs. push(key, value) runs the graph fed through one input; filter(GraphInputs) feeds several in one run. Inputs not fed in a run are empty, like dropped paths, so merges and outputs need no new rules. GraphInputs::of(keys...) declares the input types, checked when the graph is built; an undeclared input takes the type of its first reader. Unknown keys and mistyped values throw when fed. A `.` on anything but in/out is now a diagnostic instead of being ignored. Implements TODO item 2 (now item 1 after the previous removal). --- CHANGELOG.md | 13 + EXAMPLE.md | 30 ++ README.md | 40 +- TODO.md | 63 +-- apps/textPipeline/main.cpp | 20 +- .../core/filterGraph/DslFilterGraph.hpp | 419 +++++++++++++++--- .../core/filterGraph/GraphLang.hpp | 46 +- .../core/filterGraph/GraphLangHandwritten.hpp | 2 +- tests/FilterGraphTests/FilterGraphTests.cpp | 7 + tests/FilterGraphTests/GraphInputsTests.cpp | 210 +++++++++ 10 files changed, 728 insertions(+), 122 deletions(-) create mode 100644 tests/FilterGraphTests/GraphInputsTests.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index 90e8bcd..1484a8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,6 +73,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (`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. +- **Several named graph inputs** — a graph can read named inputs, + `in.`, with the new **`GraphInputs`** as its input type, mirroring + `GraphOutputs`. `DslFilterGraph::push(key, value)` runs the graph fed through + one input; `filter(GraphInputs{}.set(...).set(...))` feeds several in one + run. An input not fed in a run is empty, like a dropped path. + `GraphInputs::of(keys...)` declares the input types for the new + `DslFilterGraph(text, inputTypes)` / `validateDslGraph(text, inputTypes)` + overloads, and they are checked when the graph is built; an undeclared input + takes the type of its first reader. Feeding an unknown key throws + `std::out_of_range`, a value of the wrong type `std::invalid_argument`. + `dsl::GraphProgram::inputs` lists the inputs a program reads. - **`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. @@ -86,6 +97,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the `std::shared_ptr` for composites that forward it. ### Changed +- A `.` on anything but `in` and `out` (e.g. `msg.x`) is now a diagnostic; + it used to be ignored silently. - `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 diff --git a/EXAMPLE.md b/EXAMPLE.md index 2495988..6d60423 100644 --- a/EXAMPLE.md +++ b/EXAMPLE.md @@ -356,6 +356,35 @@ The graph's output type decides what shape is allowed: | `GraphOutputs` | one or more `-> out` / `-> out.` | all outputs; `std::nullopt` only if all were dropped | | `Void` | no outputs, only `-> end` | `Void{}` | +### Several named inputs + +The mirror image: a graph can read several named inputs, `in.`, with +`GraphInputs` as its input type. Each `push()` is one run that feeds a single +input; the stages that only an unfed input reaches are skipped, exactly as if +their path had dropped the message: + +```cpp +DslFilterGraph routes(R"dsl( + in.greeting -> Uppercase -> out.shouted + in.name -> Reverse -> out.reversed +)dsl", + GraphInputs::of("greeting", "name")); + +auto outputs = routes.push("name", std::string{"filterGraph"}); +std::cout << std::format("[inputs] shouted={} reversed={}\n", + outputs->has("shouted") ? "present" : "not fed", + *outputs->get("reversed")); +``` + +```text +[inputs] shouted=not fed reversed=hparGretlif +``` + +`GraphInputs::of` declares the input types, so they are checked when the graph +is built; it is optional, and without it each input takes the type of the +first stage that reads it. To feed several inputs in one run, pass them +together: `routes.filter(GraphInputs{}.set("greeting", a).set("name", b))`. + ## 6. Checking a graph before it runs Constructing a `DslFilterGraph` checks the whole graph before any message @@ -824,6 +853,7 @@ The complete output of `textPipeline` (sections 1–7): [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 +[inputs] shouted=not fed reversed=hparGretlif [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 [json tap] HELLO, FILTERGRAPH! diff --git a/README.md b/README.md index 2b17a22..8cfa259 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,7 @@ edge -> Stage -> edge -> Stage(key=value) -> edge | Name | Meaning | | --- | --- | | `in` | the graph's input; may only start a statement | + | `in.` | a named input, for a graph with several (see [Several named inputs](#several-named-inputs)) | | `out` | the graph's output; may only end a statement | | `out.` | a named output (several outputs are ordered as written) | | `end` | a dead end: the value is discarded (required for `Void` stages) | @@ -374,6 +375,36 @@ outputs->get(1); // outputs can also be read by position outputs->has("length"); // false if that output's path dropped the message ``` +### Several named inputs + +A graph that consumes several kinds of message reads named inputs, `in.`, +and takes `GraphInputs` as its input type, mirroring `GraphOutputs`: + +```cpp +DslFilterGraph matcher(R"dsl( + in.orders -> ParseOrder -> order + in.quotes -> ParseQuote -> quote + (order, quote) -> Match -> out +)dsl", + GraphInputs::of("orders", "quotes")); // optional: declared input types + +matcher.push("quotes", Quote{...}); // one run, fed through in.quotes only +matcher.filter(GraphInputs{}.set("orders", order).set("quotes", quote)); // one run, both inputs +``` + +- One `push` or `filter` is one run. An input not fed in that run is empty, + exactly like a dropped path: the stages reading it are skipped, and a merge + sees a hole. +- `GraphInputs::of(keys...)` declares the input types, which are then + checked like every other edge when the graph is built; a declared input the + graph never reads, and a read input that is not declared, are diagnostics. + Without a declaration, an input takes the type of the first stage that reads + it. +- Feeding an unknown key throws `std::out_of_range`; a value whose type differs + from the input's (declared or inferred) type throws `std::invalid_argument`. +- A graph uses either `in` or named inputs: `in` with `GraphInputs`, or + `in.` with any other input type, is a build-time diagnostic. + ### Checking and visualizing a graph `validateDslGraph` runs every construction check without running a message @@ -407,9 +438,10 @@ listing each stage's arguments. - Stage arguments are flat `key=value` pairs; nested objects and lists are not expressible yet. Stages that need them can be configured in JSON. -- Only the graph input type and the single `out` type are checked against the - C++ template parameters; the types inside `GraphOutputs` are checked when - they are read (`get` throws `std::bad_any_cast` on a mismatch). +- Only the graph input type, the declared named input types and the single + `out` type are checked against the C++ side; the types inside `GraphOutputs` + are checked when they are read (`get` throws `std::bad_any_cast` on a + mismatch). ## Graph context @@ -624,8 +656,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: -- **Several named graph inputs** (`in.orders`, `in.quotes`), mirroring - `GraphOutputs`. - **Stage labels and typed access** (`Summarize@stats`, `graph.stage("stats")`). - **Track which stage short-circuited**, so a caller can see where a message was dropped rather than only that it was. diff --git a/TODO.md b/TODO.md index be316b1..7832b8e 100644 --- a/TODO.md +++ b/TODO.md @@ -18,18 +18,18 @@ Examples use the generic stages of the [README](README.md) (`Parse`, | # | Item | Impact | Compatibility | Workaround today | |---|------|--------|---------------|------------------| -| 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 | +| 1 | [Stage labels and typed access](#1-stage-labels-and-typed-access) | medium | additive (DSL syntax) | side registry filled by creator lambdas | +| 2 | [Track which stage short-circuited](#2-track-which-stage-short-circuited) | medium | additive | tap the edge, or log in the stage | +| 3 | [Config-aware `registerMergeFilter`](#3-config-aware-registermergefilter) | low | additive | subclass + `FilterRegistrar` creator | +| 4 | [Injectable registry, duplicate detection](#4-injectable-registry-duplicate-detection) | low | mostly additive | unique names | +| 5 | [Documentation: 0..n outputs, large messages](#5-documentation-0n-outputs-large-messages) | doc only | — | — | +| 6 | [Known limitations to lift](#6-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`), **named merge slots** -(`(raw: msg, checked: valid) -> Merge`, `MergeStage::mergeInputNames()`), the +(`(raw: msg, checked: valid) -> Merge`, `MergeStage::mergeInputNames()`), +**several named graph inputs** (`in.`, `GraphInputs`), the **end-of-stream hook** (`MessageFilter::finish()`), all described in the README, and **examples for the newer features** — `apps/statefulPipeline` and `apps/compositePipeline`, @@ -37,42 +37,7 @@ walked through in [EXAMPLE.md](EXAMPLE.md) sections 8 and 9. --- -## 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 -every stage has to unpack it, or one "select" stage per alternative drops the -others. - -**Proposal.** A graph with named inputs, mirroring `GraphOutputs`: - -```text -in.orders -> ParseOrder -> order -in.quotes -> ParseQuote -> quote -(order, quote) -> Match -> out -``` - -```cpp -DslFilterGraph graph(text); -graph.push("quotes", Quote{...}); // runs only the stages reachable from in.quotes -``` - -Semantics: - -- One `push` is one run. -- Input edges not fed in this run are **holes**, exactly like dropped paths - today, so merges and outputs need no new rules. -- Validation checks each named input's type at construction, which needs a way - to declare them, e.g. `GraphInputs::of("orders", "quotes")`. -- `filter(GraphInputs&&)` stays available, for "push several inputs in one run". - -**Compatibility.** Additive: `DslFilterGraph` with a plain input type -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. - -## 2. Stage labels and typed access +## 1. 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. @@ -102,7 +67,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`. -## 3. Track which stage short-circuited +## 2. Track which stage short-circuited **Today.** When a graph drops a message, neither `DslFilterGraph` nor `AnyFilterChain` tells the caller *which* stage returned `std::nullopt`. @@ -123,7 +88,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. -## 4. Config-aware `registerMergeFilter` +## 3. Config-aware `registerMergeFilter` **Today.** `registerMergeFilter(name, combiner)` ignores the DSL config and **copies one combiner into every instance**. A combiner lambda that captures a @@ -153,7 +118,7 @@ one. **Workaround today.** Subclass and register with a creator lambda. -## 5. Injectable registry, duplicate detection +## 4. Injectable registry, duplicate detection **Today.** `FilterRegistry::instance()` is a process-wide singleton, and `registerFilter` **silently overwrites** an existing name. @@ -182,7 +147,7 @@ Ship it behind a transition: warn first, or add **Workaround today.** Keep names unique, and register test fakes under their own names. -## 6. Documentation: 0..n outputs, large messages +## 5. 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: @@ -199,7 +164,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. -## 7. Known limitations to lift +## 6. 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 6bcb0f3..164bed5 100644 --- a/apps/textPipeline/main.cpp +++ b/apps/textPipeline/main.cpp @@ -3,8 +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 (merges, with typed and named slots) and -// several named outputs +// with fan-out (a "tap"), fan-in (merges, with typed and named slots), +// several named outputs and several named inputs // - validateDslGraph: every problem in a broken graph, located by line:column // - JsonFilterGraph: the JSON format, which remains supported #include @@ -30,6 +30,7 @@ using filterGraph::DslFilterGraph; using filterGraph::FilterGraph; using filterGraph::FilterRegistrar; +using filterGraph::GraphInputs; using filterGraph::JsonFilterGraph; using filterGraph::MergeInputs; using filterGraph::MessageFilter; @@ -290,6 +291,21 @@ int main() outputs->has("long") ? "present" : "dropped"); } + // 4b) Several named inputs, with declared types. push() feeds one input + // and runs only the stages it reaches; the other input is empty. + { + DslFilterGraph routes(R"dsl( + in.greeting -> Uppercase -> out.shouted + in.name -> Reverse -> out.reversed + )dsl", + GraphInputs::of("greeting", "name")); + + auto outputs = routes.push("name", std::string{"filterGraph"}); + std::cout << std::format("[inputs] shouted={} reversed={}\n", + outputs->has("shouted") ? "present" : "not fed", + *outputs->get("reversed")); + } + // 5) Problems are reported before anything runs: all of them at once, each // located by line:column. Constructing a DslFilterGraph from this text // would throw a GraphError carrying the same diagnostics. diff --git a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp index b05bc89..beab82e 100644 --- a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp +++ b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp @@ -30,9 +30,17 @@ // in -> Uppercase -> shouted -> Print(prefix="[tap] ") -> end // in -> Reverse -> reversed -> MinLength(minLength=3) -> long -> Print -> out // -// Execution semantics, per message: +// A graph has one input `in`, or several named inputs `in.` (with +// GraphInputs as its input type): +// +// in.orders -> ParseOrder -> order +// in.quotes -> ParseQuote -> quote +// (order, quote) -> Match -> out +// +// Execution semantics, per message (per run, for named inputs): // - stages run in source order, except that a stage waits until every edge it // reads has been produced; +// - a named input that is not fed in a run is empty, like a dropped edge; // - fan-out: every reader of an edge gets its own copy of the value (the last // reader receives it by move); // - a stage returning std::nullopt leaves its edge empty; stages reading an @@ -135,6 +143,87 @@ class GraphOutputs std::shared_ptr mKeys; }; +namespace dsl::detail { +class GraphPlan; +} // namespace dsl::detail + +// The declared types of a graph's named inputs; see GraphInputs::of. +struct GraphInputTypes +{ + struct Input + { + std::string key; + std::type_index type; + }; + + std::vector inputs; +}; + +// The values of one run of a graph with named inputs (`in.`), the +// counterpart of GraphOutputs. An input that is not set is empty in that run: +// the stages reading it are skipped, exactly as for a dropped path. +// +// graph.filter(GraphInputs{}.set("orders", order).set("quotes", quote)); +class GraphInputs +{ +public: + // Declares the types of a graph's named inputs, one key per type, so that + // DslFilterGraph checks them when the graph is built: + // + // DslFilterGraph graph(text, GraphInputs::of("orders", "quotes")); + template + static GraphInputTypes of(Keys&&... keys) + { + static_assert(sizeof...(Types) == sizeof...(Keys), "GraphInputs::of needs one key per type"); + return GraphInputTypes{ + {GraphInputTypes::Input{std::string(std::forward(keys)), std::type_index(typeid(Types))}...}}; + } + + // Sets the value of input `key`, replacing an earlier one. + template + GraphInputs& set(std::string key, T&& value) & + { + for (auto& entry : mEntries) + { + if (entry.key == key) + { + entry.value = std::forward(value); + return *this; + } + } + mEntries.push_back({std::move(key), std::any(std::forward(value))}); + return *this; + } + + template + GraphInputs&& set(std::string key, T&& value) && + { + set(std::move(key), std::forward(value)); + return std::move(*this); + } + + bool has(std::string_view key) const + { + return std::any_of(mEntries.begin(), mEntries.end(), [&](const Entry& entry) { return entry.key == key; }); + } + + std::size_t size() const + { + return mEntries.size(); + } + +private: + friend class dsl::detail::GraphPlan; + + struct Entry + { + std::string key; + std::any value; + }; + + std::vector mEntries; +}; + // Thrown when a DSL graph cannot be built. what() lists every problem as // "line:column: message"; diagnostics() gives them individually. class GraphError : public std::runtime_error @@ -191,6 +280,10 @@ ExpectedOutputs expectedOutputs() inline std::vector runOrder(const GraphProgram& program) { std::unordered_set produced{"in"}; + for (const auto& input : program.inputs) + { + produced.insert(input.edge); + } std::vector scheduled(program.stages.size(), false); std::vector order; @@ -221,14 +314,32 @@ inline std::vector runOrder(const GraphProgram& program) return order; } -inline std::string describeEdge(const std::string& edge) +// The types known for the edges of a graph while its plan is built. A named +// input without a declared type takes the input type of the first stage that +// reads it. +struct EdgeTypes { - if (edge == "in") + std::unordered_map types; + std::unordered_map inferredFrom; // input edge -> the stage that typed it + + std::optional find(const std::string& edge) const { - return "the graph input 'in'"; + const auto type = types.find(edge); + return type != types.end() ? std::optional(type->second) : std::nullopt; } - return std::format("edge '{}'", edge); -} + + std::string describe(const std::string& edge) const + { + if (isInputEdge(edge)) + { + const auto reader = inferredFrom.find(edge); + return reader == inferredFrom.end() + ? std::format("the graph input '{}'", edge) + : std::format("the graph input '{}' (typed by its reader '{}')", edge, reader->second); + } + return std::format("edge '{}'", edge); + } +}; inline std::string joinNames(const std::vector& names) { @@ -327,11 +438,11 @@ inline std::optional> matchSlotNames(const StageNode& // 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) +inline std::vector wireMergeSlots(const StageNode& node, + const MergeSlotTypes& types, + const std::vector& names, + const EdgeTypes& edgeTypes, + std::vector& diagnostics) { if (!names.empty() && !checkDeclaredSlotNames(node, types, names, diagnostics)) { @@ -369,7 +480,7 @@ inline std::vector wireMergeSlots(const StageNode& { 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) + if (type && *type != expected) { const std::string label = names.empty() ? std::to_string(slot + 1) : std::format("'{}'", names[slot]); diagnostics.push_back({node.loc, @@ -377,8 +488,8 @@ inline std::vector wireMergeSlots(const StageNode& label, node.type, expected.name(), - describeEdge(edges[slot]), - type->second.name())}); + edgeTypes.describe(edges[slot]), + type->name())}); } } return edges; @@ -400,18 +511,23 @@ class GraphPlan GraphPlan() = default; // Builds the plan, appending every problem found (bad config, type - // mismatches along edges, misused merges/Void, output shape) to + // mismatches along edges, misused merges/Void, inputs and output shape) to // `diagnostics`. The plan may only be run if no diagnostics were added. + // `declaredInputs` are the types of a GraphInputs graph's named inputs, if + // the caller declared them. GraphPlan(const GraphProgram& program, std::type_index inputType, const ExpectedOutputs& expected, + const GraphInputTypes* declaredInputs, std::vector& diagnostics) { std::unordered_map slots; auto slotOf = [&slots](const std::string& edge) { return slots.try_emplace(edge, slots.size()).first->second; }; - slotOf("in"); // always slot 0 (kInputSlot) + + EdgeTypes edgeTypes; + bindInputs(program, inputType, declaredInputs, slotOf, edgeTypes, diagnostics); if (program.stages.empty()) { @@ -424,9 +540,7 @@ class GraphPlan diagnostics.push_back({SourceLoc{}, "some stages can never run: an input is never produced or depends on itself"}); } - auto& registry = FilterRegistry::instance(); - std::unordered_map edgeTypes; - edgeTypes.emplace("in", inputType); + auto& registry = FilterRegistry::instance(); for (std::size_t index : order) { @@ -473,15 +587,21 @@ class GraphPlan else { const std::string& edge = node.inputs.front(); - auto type = edgeTypes.find(edge); - if (type != edgeTypes.end() && type->second != filter->inputType()) + const auto type = edgeTypes.find(edge); + if (!type && isInputEdge(edge)) + { + // An undeclared named input: its first reader types it. + edgeTypes.types.emplace(edge, filter->inputType()); + edgeTypes.inferredFrom.emplace(edge, node.type); + } + else if (type && *type != filter->inputType()) { diagnostics.push_back({node.loc, std::format("stage '{}' expects input type '{}' but {} carries '{}'", node.type, filter->inputType().name(), - describeEdge(edge), - type->second.name())}); + edgeTypes.describe(edge), + type->name())}); } } @@ -495,7 +615,7 @@ class GraphPlan } else { - edgeTypes.emplace(*node.output, filter->outputType()); + edgeTypes.types.emplace(*node.output, filter->outputType()); } } @@ -523,6 +643,11 @@ class GraphPlan checkOutputs(program, expected, edgeTypes, diagnostics); + for (auto& input : mInputs) + { + input.type = edgeTypes.find(input.edge); + } + mEdgeCount = slots.size(); mReaders.assign(mEdgeCount, 0); for (const auto& stage : mStages) @@ -534,13 +659,44 @@ class GraphPlan } } + // Orders the values of a run of a GraphInputs graph for run(), checking + // their keys and, where known, their types. + std::vector bind(GraphInputs&& inputs) const + { + std::vector values(mInputs.size()); + for (auto& entry : inputs.mEntries) + { + const auto input = std::find_if(mInputs.begin(), mInputs.end(), [&](const Input& candidate) { + return candidate.key == entry.key; + }); + if (input == mInputs.end()) + { + throw std::out_of_range(std::format("DslFilterGraph: the graph has no input named '{}'", entry.key)); + } + if (input->type && entry.value.has_value() && std::type_index(entry.value.type()) != *input->type) + { + throw std::invalid_argument(std::format("DslFilterGraph: input '{}' expects '{}' but received '{}'", + entry.key, + input->type->name(), + entry.value.type().name())); + } + values[static_cast(input - mInputs.begin())] = std::move(entry.value); + } + return values; + } + // Runs one message through the graph and returns the output slots, in DSL // order; an empty slot means that output's path dropped the message. - std::vector run(std::any&& input) const + // `inputs` holds one value per graph input (see bind()); an empty one is + // not fed in this run. + std::vector run(std::vector&& inputs) const { std::vector edges(mEdgeCount); std::vector remaining = mReaders; - edges[kInputSlot] = std::move(input); + for (std::size_t i = 0; i < mInputs.size() && i < inputs.size(); ++i) + { + edges[mInputs[i].slot] = std::move(inputs[i]); + } for (const auto& stage : mStages) { @@ -613,7 +769,90 @@ class GraphPlan } private: - static constexpr std::size_t kInputSlot = 0; + struct Input + { + std::string key; // "" for the single input `in` + std::string edge; // "in" or "in." + std::size_t slot = 0; + std::optional type; // declared or inferred; nullopt if unknown + }; + + // Registers the graph's inputs: the single `in`, or the named `in.` + // of a GraphInputs graph, checked against the declared input types. + template + void bindInputs(const GraphProgram& program, + std::type_index inputType, + const GraphInputTypes* declared, + SlotOf& slotOf, + EdgeTypes& edgeTypes, + std::vector& diagnostics) + { + if (inputType != std::type_index(typeid(GraphInputs))) + { + mInputs.push_back({"", "in", slotOf("in"), inputType}); + edgeTypes.types.emplace("in", inputType); + for (const auto& input : program.inputs) + { + if (input.key) + { + diagnostics.push_back( + {input.loc, + std::format("the named input '{}' needs GraphInputs as the graph's input type", input.edge)}); + } + } + return; + } + + for (const auto& input : program.inputs) + { + if (!input.key) + { + diagnostics.push_back({input.loc, + "the graph's input type is GraphInputs, so its inputs are named: write " + "'in.' instead of 'in'"}); + continue; + } + mInputs.push_back({*input.key, input.edge, slotOf(input.edge), std::nullopt}); + if (!declared) + { + continue; + } + const auto type = std::find_if(declared->inputs.begin(), declared->inputs.end(), + [&](const GraphInputTypes::Input& d) { return d.key == *input.key; }); + if (type == declared->inputs.end()) + { + diagnostics.push_back({input.loc, + std::format("the graph reads '{}', but no input '{}' is declared", + input.edge, + *input.key)}); + } + else + { + edgeTypes.types.emplace(input.edge, type->type); + } + } + + if (!declared) + { + return; + } + std::unordered_set seen; + for (const auto& input : declared->inputs) + { + if (!seen.insert(input.key).second) + { + diagnostics.push_back({SourceLoc{}, std::format("input '{}' is declared twice", input.key)}); + continue; + } + const bool read = std::any_of(mInputs.begin(), mInputs.end(), [&](const Input& i) { return i.key == input.key; }); + if (!read) + { + diagnostics.push_back( + {SourceLoc{}, + std::format("input '{}' is declared but the graph never reads 'in.{}'", input.key, input.key)}); + } + } + } // The last reader of an edge takes the value by move; earlier readers // (fan-out) each get their own copy. @@ -626,10 +865,10 @@ class GraphPlan return edges[edge]; } - static void checkOutputs(const GraphProgram& program, - const ExpectedOutputs& expected, - const std::unordered_map& edgeTypes, - std::vector& diagnostics) + static void checkOutputs(const GraphProgram& program, + const ExpectedOutputs& expected, + const EdgeTypes& edgeTypes, + std::vector& diagnostics) { const auto& outputs = program.outputs; switch (expected.shape) @@ -652,13 +891,12 @@ class GraphPlan expected.type.name(), outputs.size())}); } - else if (auto type = edgeTypes.find(outputs[0].edge); - type != edgeTypes.end() && type->second != expected.type) + else if (auto type = edgeTypes.find(outputs[0].edge); type && *type != expected.type) { diagnostics.push_back({outputs[0].loc, std::format("the graph's output type is '{}' but 'out' receives '{}'", expected.type.name(), - type->second.name())}); + type->name())}); } break; case OutputShape::Multiple: @@ -680,6 +918,7 @@ class GraphPlan } } + std::vector mInputs; // input i is fed from run()'s inputs[i] std::vector mStages; std::size_t mEdgeCount = 0; std::vector mReaders; // per edge slot: number of stage inputs reading it @@ -693,6 +932,12 @@ class GraphPlan // already parsed GraphProgram) and exposes it as a MessageFilter, so it plugs // in anywhere a compile-time FilterGraph or a JsonFilterGraph can. // +// InputType is the type of the single input `in`, or GraphInputs for a graph +// with named inputs `in.`. Such a graph runs once per filter() (several +// inputs) or push() (one input); inputs not fed in a run are empty. Their types +// can be declared with GraphInputs::of; an undeclared input is typed by the +// first stage that reads it. +// // OutputType selects the output boundary the graph must have: // - any concrete type: exactly one `-> out` producing that type; // std::nullopt when its path drops the message; @@ -715,11 +960,75 @@ class DslFilterGraph : public MessageFilter } explicit DslFilterGraph(dsl::GraphProgram program) + : DslFilterGraph(std::move(program), nullptr) + { + } + + // A graph with named inputs whose types are declared, e.g. + // GraphInputs::of("orders", "quotes"). + DslFilterGraph(std::string_view text, const GraphInputTypes& inputs) + requires std::is_same_v + : DslFilterGraph(dsl::parseGraphProgram(text), &inputs) + { + } + + DslFilterGraph(dsl::GraphProgram program, const GraphInputTypes& inputs) + requires std::is_same_v + : DslFilterGraph(std::move(program), &inputs) + { + } + + std::optional filter(InputType&& data) override + { + std::vector inputs; + if constexpr (std::is_same_v) + { + inputs = mPlan.bind(std::move(data)); + } + else + { + inputs.push_back(std::any(std::move(data))); + } + return collect(mPlan.run(std::move(inputs))); + } + + // Runs the graph with a single named input; the others are empty in this + // run. Throws std::out_of_range for an unknown key and + // std::invalid_argument for a value of the wrong type. + template + std::optional push(std::string key, T&& value) + requires std::is_same_v + { + return filter(GraphInputs{}.set(std::move(key), std::forward(value))); + } + + void setContext(std::shared_ptr context) override + { + mPlan.setContext(context); + 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 + { + return mProgram; + } + +private: + DslFilterGraph(dsl::GraphProgram program, const GraphInputTypes* inputs) : mProgram(std::move(program)) { std::vector diagnostics = mProgram.diagnostics; mPlan = dsl::detail::GraphPlan( - mProgram, typeid(InputType), dsl::detail::expectedOutputs(), diagnostics); + mProgram, typeid(InputType), dsl::detail::expectedOutputs(), inputs, diagnostics); if (!diagnostics.empty()) { dsl::detail::sortDiagnostics(diagnostics); @@ -729,10 +1038,8 @@ class DslFilterGraph : public MessageFilter DslFilterGraph::setContext(this->sharedContext()); } - std::optional filter(InputType&& data) override + std::optional collect(std::vector&& outputs) const { - std::vector outputs = mPlan.run(std::any(std::move(data))); - if constexpr (std::is_same_v) { return Void{}; @@ -758,27 +1065,6 @@ class DslFilterGraph : public MessageFilter } } - void setContext(std::shared_ptr context) override - { - mPlan.setContext(context); - 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 - { - return mProgram; - } - -private: dsl::GraphProgram mProgram; dsl::detail::GraphPlan mPlan; }; @@ -792,7 +1078,20 @@ std::vector validateDslGraph(std::string_view text) const dsl::GraphProgram program = dsl::parseGraphProgram(text); std::vector diagnostics = program.diagnostics; [[maybe_unused]] const dsl::detail::GraphPlan plan( - program, typeid(InputType), dsl::detail::expectedOutputs(), diagnostics); + program, typeid(InputType), dsl::detail::expectedOutputs(), nullptr, diagnostics); + dsl::detail::sortDiagnostics(diagnostics); + return diagnostics; +} + +// The same for a graph with named inputs whose types are declared. +template + requires std::is_same_v +std::vector validateDslGraph(std::string_view text, const GraphInputTypes& inputs) +{ + const dsl::GraphProgram program = dsl::parseGraphProgram(text); + std::vector diagnostics = program.diagnostics; + [[maybe_unused]] const dsl::detail::GraphPlan plan( + program, typeid(InputType), dsl::detail::expectedOutputs(), &inputs, diagnostics); dsl::detail::sortDiagnostics(diagnostics); return diagnostics; } diff --git a/libs/filterGraph/core/filterGraph/GraphLang.hpp b/libs/filterGraph/core/filterGraph/GraphLang.hpp index 263af1f..18046ae 100644 --- a/libs/filterGraph/core/filterGraph/GraphLang.hpp +++ b/libs/filterGraph/core/filterGraph/GraphLang.hpp @@ -37,8 +37,8 @@ // msg -> Store -> end // // - even positions are edges (identifiers) or the reserved boundary nodes -// `in` (graph input), `out` / `out.` (results), `end` (dead-end: the -// value is discarded); +// `in` / `in.` (the graph input, or one of several named inputs), +// `out` / `out.` (results), `end` (dead-end: the value is discarded); // - odd positions are stage applications: a REGISTERED filter/merge name, // optionally with config args `Name(k=v, k2="s", k3=true)`; // - fan-out = reuse an edge name as a source in several statements; @@ -79,6 +79,15 @@ struct StageNode SourceLoc loc; }; +// One graph input the program reads: `in`, or a named `in.`, located at +// its first use. +struct InputBinding +{ + std::string edge; // "in" or "in." + std::optional key; // name from `in.` + SourceLoc loc; +}; + // One `-> out` (or `-> out.`), in DSL order (index is positional). struct OutputBinding { @@ -91,6 +100,7 @@ struct OutputBinding struct GraphProgram { std::vector stages; + std::vector inputs; // distinct graph inputs, in order of first use std::vector outputs; std::vector deadEnds; // edges routed to `end` std::vector diagnostics; @@ -136,6 +146,12 @@ inline bool isReserved(std::string_view name) return name == "in" || name == "out" || name == "end"; } +// The edge a graph input arrives on: `in`, or `in.` for a named input. +inline bool isInputEdge(std::string_view edge) +{ + return edge == "in" || edge.starts_with("in."); +} + // ------------------------- shared lexical helpers ------------------------ // // The parser below and the deprecated hand-written parser @@ -307,7 +323,7 @@ struct Term Kind kind = Kind::Edge; std::string name; // edge/stage/boundary name - std::optional key; // for `out.` + std::optional key; // for `in.` / `out.` nlohmann::json config; // for Stage std::vector edges; // for Group std::vector slotNames; // for Group: per edge, from `name: edge`; "" when unnamed @@ -407,6 +423,15 @@ inline void buildStatement(std::vector& terms, return; } } + + // Only the graph's inputs and outputs are keyed. + if (terms[i].key && terms[i].name != "in" && terms[i].name != "out") + { + diags.push_back({terms[i].loc, + std::format("only 'in' and 'out' take a '.', not '{}.{}'", terms[i].name, + *terms[i].key)}); + return; + } } // Emit one stage node per stage term, wiring input(s) from the previous @@ -429,6 +454,17 @@ inline void buildStatement(std::vector& terms, node.slotNames = source.slotNames; } } + else if (source.name == "in") + { + const std::string edge = source.key ? "in." + *source.key : std::string{"in"}; + const bool seen = std::any_of(program.inputs.begin(), program.inputs.end(), + [&](const InputBinding& input) { return input.edge == edge; }); + if (!seen) + { + program.inputs.push_back({edge, source.key, source.loc}); + } + node.inputs = {edge}; + } else { node.inputs = {source.name}; @@ -492,7 +528,7 @@ inline void validateProgram(GraphProgram& program) { consumed.insert(input); const bool produced = producerCount.find(input) != producerCount.end(); - if (input != "in" && !produced) + if (!isInputEdge(input) && !produced) { program.diagnostics.push_back( {stage.loc, std::format("edge '{}' is used but never produced", input)}); @@ -862,7 +898,7 @@ inline bool parseStatementLine(std::string_view lineText, term.loc = locOf(sc.position()); term.name = captureIdent(); - // Optional `.key` (only meaningful for `out`). + // Optional `.key` (only meaningful for `in` and `out`). skipBlank(); if (sc.branch(ld::lit_c<'.'>)) { diff --git a/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp b/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp index 21b15f6..6be002b 100644 --- a/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp +++ b/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp @@ -349,7 +349,7 @@ class Parser term.name = peek().text; next(); - // Optional `.key` (currently only meaningful for `out`). + // Optional `.key` (only meaningful for `in` and `out`). if (peek().kind == TokenKind::Dot) { next(); diff --git a/tests/FilterGraphTests/FilterGraphTests.cpp b/tests/FilterGraphTests/FilterGraphTests.cpp index 67f9cdc..72b03f4 100644 --- a/tests/FilterGraphTests/FilterGraphTests.cpp +++ b/tests/FilterGraphTests/FilterGraphTests.cpp @@ -945,6 +945,11 @@ std::string describeProgram(const dsl::GraphProgram& program) text += std::format("output {} key={} index={} @{}:{}\n", output.edge, output.key.value_or(""), output.index, output.loc.line, output.loc.column); } + for (const auto& input : program.inputs) + { + text += std::format("input {} key={} @{}:{}\n", input.edge, input.key.value_or(""), input.loc.line, + input.loc.column); + } text += std::format("deadEnds={}\n", program.deadEnds.size()); text += dsl::formatDiagnostics(program.diagnostics); return text; @@ -1002,6 +1007,8 @@ TEST_CASE("Both parsers report the same located syntax errors", "[GraphDslErrors {"(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 ':'"}, + {"in -> RunDouble -> q.x -> RunDouble -> out", "1:20: only 'in' and 'out' take a '.', not 'q.x'"}, + {"in. -> RunDouble -> out", "1:5: expected a key name after '.' but found '->'"}, }; for (const auto& [source, expected] : cases) diff --git a/tests/FilterGraphTests/GraphInputsTests.cpp b/tests/FilterGraphTests/GraphInputsTests.cpp new file mode 100644 index 0000000..845761a --- /dev/null +++ b/tests/FilterGraphTests/GraphInputsTests.cpp @@ -0,0 +1,210 @@ +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace filterGraph; + +namespace { + +struct Order +{ + int quantity = 0; +}; + +struct Quote +{ + double price = 0.0; +}; + +class ParseOrderFilter : public MessageFilter +{ +public: + std::optional filter(Order&& order) override + { + return order.quantity; + } +}; + +class ParseQuoteFilter : public MessageFilter +{ +public: + std::optional filter(Quote&& quote) override + { + return quote.price; + } +}; + +class IntToStringFilter : public MessageFilter +{ +public: + std::optional filter(int&& value) override + { + return std::to_string(value); + } +}; + +// "@", with "-" for a hole. +class MatchMerge : public TypedMergeFilter +{ +public: + std::optional merge(std::optional&& quantity, std::optional&& price) override + { + return std::format("{}@{}", quantity ? std::to_string(*quantity) : "-", price ? std::format("{}", *price) : "-"); + } +}; + +static FilterRegistrar registerParseOrder("InParseOrder"); +static FilterRegistrar registerParseQuote("InParseQuote"); +static FilterRegistrar registerIntToString("InIntToString"); +static FilterRegistrar registerMatch("InMatch"); + +const char* kMatchGraph = R"( + in.orders -> InParseOrder -> quantity + in.quotes -> InParseQuote -> price + (quantity, price) -> InMatch -> out +)"; + +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("push runs only the stages reachable from one named input", "[GraphInputs]") +{ + DslFilterGraph graph(kMatchGraph); + + REQUIRE(graph.push("orders", Order{3}).value() == "3@-"); + REQUIRE(graph.push("quotes", Quote{1.5}).value() == "-@1.5"); +} + +TEST_CASE("filter feeds several named inputs in one run", "[GraphInputs]") +{ + DslFilterGraph graph(kMatchGraph); + + REQUIRE(graph.filter(GraphInputs{}.set("orders", Order{2}).set("quotes", Quote{0.5})).value() == "2@0.5"); + REQUIRE_FALSE(graph.filter(GraphInputs{}).has_value()); // nothing fed: every path is empty +} + +TEST_CASE("Named inputs feed separate outputs", "[GraphInputs]") +{ + DslFilterGraph graph(R"( + in.orders -> InParseOrder -> out.quantity + in.quotes -> InParseQuote -> out.price + )"); + + auto outputs = graph.push("quotes", Quote{2.0}); + REQUIRE(outputs.has_value()); + REQUIRE_FALSE(outputs->has("quantity")); + REQUIRE(outputs->get("price").value() == 2.0); +} + +TEST_CASE("Declared input types are checked when the graph is built", "[GraphInputs]") +{ + const auto declared = GraphInputs::of("orders", "quotes"); + DslFilterGraph graph(kMatchGraph, declared); + REQUIRE(graph.push("orders", Order{1}).value() == "1@-"); + + const auto swapped = validateDslGraph(kMatchGraph, GraphInputs::of("orders", "quotes")); + REQUIRE(swapped.size() == 2); + REQUIRE(swapped[0].message.starts_with("stage 'InParseOrder' expects input type")); + REQUIRE(swapped[0].message.find("the graph input 'in.orders' carries") != std::string::npos); + REQUIRE(swapped[0].loc.line == 2); +} + +TEST_CASE("Declared and read inputs must agree", "[GraphInputs]") +{ + const auto undeclared = + validateDslGraph(kMatchGraph, GraphInputs::of("orders")); + REQUIRE(undeclared.size() == 1); + REQUIRE(undeclared[0].message == "the graph reads 'in.quotes', but no input 'quotes' is declared"); + REQUIRE(undeclared[0].loc.line == 3); + + const auto unread = validateDslGraph( + kMatchGraph, GraphInputs::of("orders", "quotes", "extra")); + REQUIRE(unread.size() == 1); + REQUIRE(unread[0].message == "input 'extra' is declared but the graph never reads 'in.extra'"); +} + +TEST_CASE("An undeclared input is typed by its first reader", "[GraphInputs]") +{ + const auto diagnostics = validateDslGraph(R"( + in.orders -> InParseOrder -> out.a + in.orders -> InIntToString -> out.b + )"); + + REQUIRE(diagnostics.size() == 1); + REQUIRE(diagnostics[0].message.starts_with("stage 'InIntToString' expects input type")); + REQUIRE(diagnostics[0].message.find("the graph input 'in.orders' (typed by its reader 'InParseOrder')") + != std::string::npos); +} + +TEST_CASE("Feeding an unknown or mistyped input throws", "[GraphInputs]") +{ + DslFilterGraph graph(kMatchGraph); + + REQUIRE_THROWS_AS(graph.push("trades", Order{1}), std::out_of_range); + REQUIRE_THROWS_AS(graph.push("orders", Quote{1.0}), std::invalid_argument); + REQUIRE_THROWS_AS(graph.filter(GraphInputs{}.set("orders", 7)), std::invalid_argument); +} + +TEST_CASE("Input shape must match the graph's input type", "[GraphInputs]") +{ + const auto plainInNamedGraph = validateDslGraph("in -> InParseOrder -> q -> InIntToString -> out"); + REQUIRE(plainInNamedGraph.size() == 1); + REQUIRE(plainInNamedGraph[0].message + == "the graph's input type is GraphInputs, so its inputs are named: write 'in.' instead of 'in'"); + + const auto namedInPlainGraph = validateDslGraph("in.orders -> InParseOrder -> q -> InIntToString -> out"); + REQUIRE(namedInPlainGraph.size() == 1); + REQUIRE(namedInPlainGraph[0].message == "the named input 'in.orders' needs GraphInputs as the graph's input type"); + REQUIRE(namedInPlainGraph[0].loc.column == 1); +} + +TEST_CASE("Only in and out take a key", "[GraphInputs]") +{ + const auto program = dsl::parseGraphProgram("in -> InParseOrder -> q.x -> InIntToString -> out"); + REQUIRE(program.diagnostics.size() == 1); + REQUIRE(program.diagnostics[0].message == "only 'in' and 'out' take a '.', not 'q.x'"); +} + +TEST_CASE("The parsed program lists its inputs", "[GraphInputs]") +{ + const auto program = dsl::parseGraphProgram("in.orders -> InParseOrder -> a\n" + "in.quotes -> InParseQuote -> b\n" + "in.orders -> InParseOrder -> c\n" + "(a, b, c) -> InMatch -> out\n"); + REQUIRE(program.inputs.size() == 2); + REQUIRE(program.inputs[0].edge == "in.orders"); + REQUIRE(program.inputs[0].key == "orders"); + REQUIRE(program.inputs[1].loc.line == 2); + REQUIRE(program.stages[0].inputs.front() == "in.orders"); + + const auto plain = dsl::parseGraphProgram("in -> InParseOrder -> out"); + REQUIRE(plain.inputs.size() == 1); + REQUIRE_FALSE(plain.inputs[0].key.has_value()); +} + +TEST_CASE("A graph with named inputs nests as a stage", "[GraphInputs]") +{ + FilterRegistrar> registrar( + "InNestedMatch", + [](const nlohmann::json&) { return std::make_shared>(kMatchGraph); }); + + DslFilterGraph outer("in.both -> InNestedMatch -> out", + GraphInputs::of("both")); + REQUIRE(outer.push("both", GraphInputs{}.set("orders", Order{4}).set("quotes", Quote{0.25})).value() == "4@0.25"); +}