diff --git a/CHANGELOG.md b/CHANGELOG.md index 4169c97..75e5c35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`DslFilterGraph`** (`DslFilterGraph.hpp`) — runs + graphs described in the text DSL, making it a first-class way to describe + runtime graphs. Executes named-edge graphs with fan-out (every reader gets a + copy), fan-in merges (holes for dropped paths), drop propagation and dead + ends, and exposes the graph as a `MessageFilter`, so it can be nested or + registered as a stage. Construction instantiates and type-checks the whole + graph and throws a `GraphError` listing every located problem. +- **`GraphOutputs`** — the ordered, optionally keyed results of a graph with + several outputs (`get` / `take` / `has`). +- **`MergeFilter` / `registerMergeFilter`** (`MergeFilter.hpp`) — + fan-in stages for `(a, b) -> Merge`, using the same combiner signature as + `JoinFilter`. +- **`validateDslGraph(text)`** — every build-time diagnostic without + throwing; `dsl::formatDiagnostic(s)` formats them as `line:column: message`. +- `dsl::StageNode::fanIn` records whether a stage's inputs came from a group. + +### Changed +- README, EXAMPLE.md and `apps/textPipeline` now describe runtime graphs in the + DSL first; the JSON format is documented as a supported alternative. +- `dsl::toMermaid` now uses generated node ids (instead of edge names such as + `end` or the internal `$out0`), labels outputs as `out` / `out.`, gives + each dead end its own node, and lists stage arguments. +- DSL stages written without arguments now receive an empty config object + instead of `null`, matching JSON stages without `"config"`. +- `dsl::parseGraphProgram` now uses the lexy-based parser, which moved into + `GraphLang.hpp`. It reports located, descriptive diagnostics (the lexy parser + used to report every problem as a bare "syntax error" at column 1), gives + stages their real columns, and supports decimal numbers, string escapes and + `#` inside strings. `GraphLangLexy.hpp` / `parseGraphProgramLexy` remain as + an alias. +- DSL syntax errors say what was expected and what was found; a syntax error + ends its statement instead of producing follow-on errors; an unterminated + string no longer swallows the following lines; malformed or out-of-range + numbers and empty fan-in groups `()` are reported; messages say "argument" + instead of "config"; diagnostics are sorted by location. + +### Deprecated +- The hand-written DSL parser. It moved to `GraphLangHandwritten.hpp` as + `dsl::parseGraphProgramHandwritten`, marked `[[deprecated]]`, and will be + removed in a future release. It produces the same results as + `dsl::parseGraphProgram`; the tests check the two against each other. + ## [0.2.0] - 2026-09-15 ### Added diff --git a/EXAMPLE.md b/EXAMPLE.md index 23c5834..beedf44 100644 --- a/EXAMPLE.md +++ b/EXAMPLE.md @@ -3,8 +3,9 @@ This walkthrough follows the runnable sample in [`apps/textPipeline/main.cpp`](apps/textPipeline/main.cpp), a small text pipeline that exercises every core building block: a **compile-time** -`FilterGraph`, and a **runtime**, JSON-configured `JsonFilterGraph` with a -parallel `FanoutFilter` branch and a scatter-gather `JoinFilter`. +`FilterGraph`, then **runtime graphs described in the text DSL** — a tap +(fan-out), a merge (fan-in), several named outputs and up-front diagnostics — +and finally the same kind of pipeline in the **JSON format**. ## Core concept: a chain of stages @@ -13,22 +14,18 @@ Every stage is a `MessageFilter`. It consumes an feeds the next stage's `InType`, forming a chain. Returning `std::nullopt` **short-circuits** the chain: no later stage runs. -The short-circuit is handled by the container (`FilterGraph` / -`AnyFilterChain`), not by the stages. As soon as a stage returns -`std::nullopt`, the chain stops and propagates `std::nullopt` out as its own -result. A downstream stage only ever receives a real, unwrapped value, so -stages never have to handle an empty optional as input — their only -responsibility is deciding whether to emit `std::nullopt` themselves. Within a -`FanoutFilter` branch the same rule applies, but the termination is confined to -that branch and never reaches the main path. - -A stage's last `OutputType` is normally the graph's own result: a path -typically ends in a stage that produces a real value (e.g. `JsonFilterGraph` yields an `Out`). A stage may instead declare `MessageFilter` to mark the path as a pure side-effect *sink* — it deliberately produces -no consumable output, so the path ends there. This is distinct from -`std::nullopt`, which means a single message was dropped or could not be -processed. A `Void` stage must be the last stage in a path. +The short-circuit is handled by the graph (`FilterGraph`, `DslFilterGraph`, +`JsonFilterGraph`), not by the stages. As soon as a stage returns +`std::nullopt`, everything downstream of it is skipped. A downstream stage only +ever receives a real, unwrapped value, so stages never have to handle an empty +optional as input — their only responsibility is deciding whether to emit +`std::nullopt` themselves. + +A path normally ends in a stage that produces a real value (the graph's +result). A stage may instead declare `MessageFilter` to mark +the path as a pure side-effect *sink* — it deliberately produces no consumable +output. This is distinct from `std::nullopt`, which means a single message was +dropped or could not be processed. ```mermaid flowchart LR @@ -39,8 +36,8 @@ flowchart LR B -. "nullopt" .-> Drop ``` -The whole chain is *itself* a `MessageFilter`, so a graph can -be nested inside another graph anywhere a single stage is expected. +Every graph is *itself* a `MessageFilter`, so a graph can be nested +inside another graph anywhere a single stage is expected. ## 1. Compile-time pipeline @@ -72,207 +69,324 @@ Output: [compile-time] !HPARGRETLIF ,OLLEH ``` -## 2. Runtime, JSON-configured pipeline with a fan-out branch +## 2. Registering stages by name -The same stages are registered by name (via `FilterRegistrar` / -`registerFanoutFilter`) and assembled at runtime from JSON. A `FanoutFilter` -duplicates the incoming message to one or more independent branches (each a -full multi-stage path), then passes the **original** message through unchanged -to the rest of the main path. +A runtime graph refers to stages by name, so each stage is registered once +with `FilterRegistrar`. Stages that take arguments supply a creator that reads +them from a JSON object: -```json -[ - { "type": "Fanout", "config": { "branches": [ - [ { "type": "Uppercase" }, { "type": "Print", "config": { "prefix": "[branch] " } } ] - ] } }, - { "type": "Reverse" }, - { "type": "MinLength", "config": { "minLength": 3 } }, - { "type": "Print", "config": { "prefix": "[main] " } } -] +```cpp +static FilterRegistrar registerUppercase("Uppercase"); +static FilterRegistrar registerReverse("Reverse"); +static FilterRegistrar registerLength("Length"); // string -> std::size_t + +// Stage arguments, e.g. `MinLength(minLength=3)`, arrive as a JSON object. +static FilterRegistrar registerMinLength( + "MinLength", + [](const nlohmann::json& config) { + return std::make_shared(config.at("minLength").get()); + }); + +static FilterRegistrar registerPrint( + "Print", + [](const nlohmann::json& config) { + return std::make_shared(config.value("prefix", std::string{})); + }); ``` +The same registry serves both the DSL and the JSON format. + +## 3. A runtime graph in the DSL, with a tap + +A DSL graph is a list of statements. Each statement alternates **edges** (named +values) and **stages**, joined by `->`. `in` is the graph's input, `out` its +output, and `end` discards a value. + +```text +in -> Uppercase -> shouted -> Print(prefix="[tap] ") -> end +in -> Reverse -> reversed -> MinLength(minLength=3) -> long -> Print(prefix="[main] ") -> out +``` + +Both statements read `in`, which **fans the message out**: each reader gets its +own copy. The first statement is a *tap* — it uppercases and prints its copy, +then discards the result at `end`. The second is the main path: it reverses the +message, drops it if it is too short, and prints it as the graph's output. + ```cpp -JsonFilterGraph pipeline(pipelineConfig); +DslFilterGraph pipeline(R"dsl( + in -> Uppercase -> shouted -> Print(prefix="[tap] ") -> end + in -> Reverse -> reversed -> MinLength(minLength=3) -> long -> Print(prefix="[main] ") -> out +)dsl"); + pipeline.filter(std::string{"Hello, filterGraph!"}); -pipeline.filter(std::string{"ab"}); // dropped by MinLength on the main path +pipeline.filter(std::string{"ab"}); // tapped, then dropped by MinLength on the main path ``` +> The raw string uses a custom delimiter (`R"dsl(...)dsl"`) because the graph +> text contains `")`, which would end a plain `R"(...)"` literal. + ```mermaid flowchart LR - In(["std::string
"Hello, filterGraph!""]) --> F{{"FanoutFilter<string>"}} - - subgraph branch ["side branch (copy, output discarded)"] - direction LR - B1["Uppercase"] --> B2["Print
prefix=[branch]"] - end - - F -. "copy" .-> B1 - F ==>|"original, unchanged"| R["Reverse"] - R -->|"!HPARGRETLIF ,OLLEH"| M["MinLength
minLength=3"] - M ==>|"len ≥ 3"| P["Print
prefix=[main]"] - M -. "len < 3" .-> Drop[["dropped"]] - P --> Out(["stdout"]) + In(["in
"Hello, filterGraph!""]) + In -. "copy" .-> U["Uppercase"] + U -->|"shouted"| PT["Print
prefix=[tap]"] + PT --> End[["end
(discarded)"]] + In ==>|"last reader: moved"| R["Reverse"] + R -->|"reversed"| M["MinLength
minLength=3"] + M ==>|"long (len ≥ 3)"| PM["Print
prefix=[main]"] + M -. "nullopt (len < 3)" .-> Drop[["dropped"]] + PM -->|"out: int status"| Out(["stdout"]) ``` Output for the two inputs: ```text -[branch] HELLO, FILTERGRAPH! -[main] !HPARGRETLIF ,OLLEH +[tap] HELLO, FILTERGRAPH! +[main] !hparGretlif ,olleH +[tap] AB ``` -The second call, `"ab"`, is copied to the branch (which uppercases and prints -it), but on the main path `MinLength` returns `std::nullopt`, so it is dropped -before the final `[main]` print. - -> Note: branch order vs. main-path order is a side-effect of how the fanout -> runs its branches before returning the original message; the branch print -> happens first. - -### Which branch contributes to the output? - -None of them. A `FanoutFilter` is a pure *tap*: it hands each branch its own -copy of the message, runs every branch to completion, and **discards each -branch's result**. Whatever the branches compute (or whether a branch -short-circuits with `std::nullopt`) never affects the main path — the value the -fanout forwards downstream is always the unchanged original input. This holds -regardless of how many branches there are or in what order they run; branches -matter only for their side effects (logging, forwarding, metrics). If you need -a branch's result to feed the output, that requires a separate merge/reduce -stage — `FanoutFilter` deliberately does not merge. - -## 3. Join: scatter-gather (the mirror of fan-out) - -Where `FanoutFilter` splits 1→N and discards branch results, `JoinFilter` -gathers N→1: it scatters the **same** input (a copy) through N independent, -multi-stage paths, then hands their outputs to a **combiner** that produces a -single result. Paths may emit different types, so their results are gathered -type-erased (one `std::any` slot per path, in order). The combiner is supplied -in C++ (via `registerJoinFilter`), since it cannot be expressed in -JSON; the paths themselves come from configuration. - -A path that short-circuits with `std::nullopt` leaves an **empty slot** (a -"hole"): the combiner sees whatever arrived and decides how to treat the -missing paths. A `Void`-terminated path produces no value to gather and is -rejected at construction. - -```json -[ - { "type": "Join", "config": { "paths": [ - [ { "type": "Uppercase" } ], - [ { "type": "Reverse" } ], - [ { "type": "MinLength", "config": { "minLength": 100 } } ] - ] } }, - { "type": "Print", "config": { "prefix": "[join] " } } -] -``` +A few things to notice: + +- **Stages run in the order they are written**, so the tap prints first. (A + stage whose input comes from a later statement simply waits until that edge + has been produced.) +- **The main path sees the original message**, not the tap's uppercased copy: + the tap works on its own copy, so `[main]` prints the reversed, mixed-case + text. +- **`"ab"` is tapped, then dropped.** `MinLength` returns `std::nullopt`, which + leaves the edge `long` empty, so the final `Print` is skipped and the graph + returns `std::nullopt` for that message. -The combiner concatenates the present outputs with `" | "` and reports how many -holes it saw: +## 4. Fan-in: merging paths + +A **merge** stage gathers several edges into one value. Writing +`(a, b, c) -> Merge` passes the merge one `MergeInputs` slot +(`std::vector`) per edge, in the order listed. The combiner lives in +C++ and is registered with `registerMergeFilter`: ```cpp -registerJoinFilter( - "Join", - [](std::vector&& outputs) -> std::optional { +registerMergeFilter( + "Concat", + [](MergeInputs&& inputs) -> std::optional { std::string joined; std::size_t holes = 0; - for (auto& out : outputs) + for (auto& input : inputs) { - if (!out.has_value()) { ++holes; continue; } // dropped path -> hole + if (!input.has_value()) { ++holes; continue; } // dropped path -> hole if (!joined.empty()) { joined += " | "; } - joined += std::any_cast(out); + joined += std::any_cast(input); } return joined + std::format(" ({} hole{})", holes, holes == 1 ? "" : "s"); }); ``` -```mermaid -flowchart LR - In(["std::string
"Hello, filterGraph!""]) --> J{{"JoinFilter<string, string>"}} +Three paths read the same input, and the merge combines them: - J -. "copy" .-> P1["Uppercase"] - J -. "copy" .-> P2["Reverse"] - J -. "copy" .-> P3["MinLength
minLength=100"] +```cpp +DslFilterGraph pipeline(R"dsl( + in -> Uppercase -> upper + in -> Reverse -> reversed + in -> MinLength(minLength=100) -> long + (upper, reversed, long) -> Concat -> joined -> Print(prefix="[merge] ") -> out +)dsl"); - P1 -->|"HELLO, FILTERGRAPH!"| C{{"combiner
(gather N slots)"}} - P2 -->|"!hparGretlif ,olleH"| C - P3 -. "nullopt = hole" .-> C +pipeline.filter(std::string{"Hello, filterGraph!"}); +``` - C -->|"joined string"| Pr["Print
prefix=[join]"] - Pr --> Out(["stdout"]) +```mermaid +flowchart LR + In(["in"]) -. "copy" .-> U["Uppercase"] + In -. "copy" .-> R["Reverse"] + In ==>|"moved"| M["MinLength
minLength=100"] + U -->|"upper"| C{{"Concat
(merge)"}} + R -->|"reversed"| C + M -. "long: empty = hole" .-> C + C -->|"joined"| P["Print
prefix=[merge]"] + P --> Out(["out"]) ``` Output: ```text -[join] HELLO, FILTERGRAPH! | !hparGretlif ,olleH (1 hole) +[merge] HELLO, FILTERGRAPH! | !hparGretlif ,olleH (1 hole) ``` -The `MinLength=100` path drops the 19-character input, so its slot is a hole; -the combiner skips it, joins the two transformed outputs, and notes the single -missing path. +The `MinLength=100` path drops the 19-character input, so its slot is a +**hole**; the combiner skips it, joins the two transformed outputs, and notes +the missing path. A merge is skipped only when *every* one of its edges is +empty — then there is nothing to combine, and the drop propagates. + +## 5. Several named outputs -## How construction-time validation works +A graph can have more than one output. `out.` names each one, and the +outputs may have different types. With `GraphOutputs` as the graph's output +type (the default), `filter()` returns all of them: -`AnyFilterChain` resolves each stage by name through the global -`FilterRegistry`, then checks that consecutive stages' erased types line up -(`outputType()` of one must equal `inputType()` of the next). `JsonFilterGraph` -additionally validates that the chain's first `inputType()` and last -`outputType()` match its own `InputType`/`OutputType`. Any mismatch throws at -construction time — the pipeline fails fast, before a single message flows. +```cpp +DslFilterGraph analysis(R"dsl( + in -> Uppercase -> out.upper + in -> Length -> out.length + in -> MinLength(minLength=100) -> out.long +)dsl"); + +auto outputs = analysis.filter(std::string{"Hello, filterGraph!"}); +std::cout << std::format("[outputs] upper={} length={} long={}\n", + *outputs->get("upper"), + *outputs->get("length"), + outputs->has("long") ? "present" : "dropped"); +``` ```mermaid -flowchart TD - Json["JSON stage array"] --> Chain["AnyFilterChain"] - Chain -->|"lookup by type name"| Reg[("FilterRegistry")] - Reg -->|"creator(config)"| Stage["AnyMessageFilter
(type-erased stage)"] - Chain -->|"outputType == next inputType?"| Check{"types line up?"} - Check -->|"no"| Throw[["throw std::runtime_error"]] - Check -->|"yes"| Ok(["validated chain"]) +flowchart LR + In(["in"]) --> U["Uppercase"] --> O1(["out.upper
std::string"]) + In --> L["Length"] --> O2(["out.length
std::size_t"]) + In --> M["MinLength
minLength=100"] -. "nullopt" .-> O3(["out.long
(empty)"]) +``` + +Output: + +```text +[outputs] upper=HELLO, FILTERGRAPH! length=19 long=dropped ``` -## Pre-flight validation: collect every problem at once +Outputs can also be read by position (`get(1)`) or moved out (`take`). +The graph's output type decides what shape is allowed: + +| `OutputType` | Graph must have | `filter()` returns | +| --- | --- | --- | +| a concrete type (sections 3 and 4 use `int`) | exactly one `-> out` of that type | the value, or `std::nullopt` if dropped | +| `GraphOutputs` | one or more `-> out` / `-> out.` | all outputs; `std::nullopt` only if all were dropped | +| `Void` | no outputs, only `-> end` | `Void{}` | + +## 6. Checking a graph before it runs -Construction fails fast — it throws on the *first* problem, with no location. -When you are authoring a config, `validateGraph` is the friendlier counterpart: -it walks the same JSON **without running any messages** and returns **all** the -problems it finds, each carrying a JSON pointer to the offending node. +Constructing a `DslFilterGraph` checks the whole graph before any message +flows, and reports **every** problem at once, each located by `line:column`. +`validateDslGraph` runs the same checks and returns the diagnostics instead of +throwing: ```cpp -#include +const auto diagnostics = validateDslGraph( + "in -> Uppercas -> shouted -> Print -> out\n" + "in -> MinLength -> long -> Print -> end\n"); -for (const auto& d : filterGraph::validateGraph(config)) +for (const auto& diagnostic : diagnostics) { - std::cout << d.pointer << ": " << d.message << '\n'; + std::cout << "[check] " << dsl::formatDiagnostic(diagnostic) << '\n'; } ``` -Given a config with a typo and a nested mistake: +Output: -```json -[ - { "type": "Uppercas" }, - { "type": "Join", "config": { "paths": [ - [ { "type": "Revrse" } ] - ] } } -] +```text +[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 ``` -it reports both, located: +Constructing a `DslFilterGraph` from the same text throws a `GraphError` whose +`what()` lists the same lines and whose `diagnostics()` returns them. + +```mermaid +flowchart TD + Text["DSL text"] --> Parse["dsl::parseGraphProgram
syntax, unknown names, edge wiring, cycles"] + Parse --> Program["GraphProgram
(stages + named edges)"] + Program --> Build["DslFilterGraph construction"] + Build -->|"create(name, arguments)"| Reg[("FilterRegistry")] + Reg --> Check{"config ok? types match along every edge?
merges, Void and outputs used correctly?"} + Check -->|"no"| Throw[["throw GraphError
(all diagnostics)"]] + Check -->|"yes"| Ok(["runnable graph"]) +``` + +What gets checked: + +- **Syntax** — a statement must alternate edges and stages; `in` only starts a + statement, `out`/`end` only end one. +- **Names** — unknown stages, with a "did you mean" suggestion. +- **Wiring** — an edge read but never written, written twice, or written but + never read; duplicate output keys; cycles. +- **Arguments** — every stage is constructed, so bad or missing arguments are + reported. +- **Types** — every edge's type must match what its reader expects, starting + from the graph's `InputType` at `in`. A mismatch names the stage, the edge and + both types (type names come from `typeid`, so their spelling depends on the + compiler). +- **Fan-in and `Void`** — groups must feed merge stages, merge stages need a + group, and a stage producing `Void` must route to `end`. +- **Outputs** — the number and type of outputs must fit the graph's + `OutputType`. + +### Visualizing a graph + +`dsl::toMermaid` renders a parsed graph as a Mermaid flowchart. For the graph in +section 3, `dsl::toMermaid(pipeline.program())` produces: + +```mermaid +flowchart LR + s0["Uppercase"] + e0(["in"]) + e1(["shouted"]) + s1["Print
prefix=[tap] "] + d0[["end"]] + s2["Reverse"] + e2(["reversed"]) + s3["MinLength
minLength=3"] + e3(["long"]) + s4["Print
prefix=[main] "] + e4(["out"]) + e0 --> s0 + s0 --> e1 + e1 --> s1 + s1 --> d0 + e0 --> s2 + s2 --> e2 + e2 --> s3 + s3 --> e3 + e3 --> s4 + s4 --> e4 +``` + +## 7. The same pipeline in JSON + +The JSON format remains supported and uses the same registered stages. JSON +chains are linear, so branching needs composite stages: here a `Fanout` +(registered with `registerFanoutFilter("Fanout")`) runs a side +branch on a copy and passes the original on — the tap from section 3: + +```cpp +auto pipelineConfig = nlohmann::json::parse(R"json([ + { "type": "Fanout", "config": { "branches": [ + [ { "type": "Uppercase" }, { "type": "Print", "config": { "prefix": "[json tap] " } } ] + ] } }, + { "type": "Reverse" }, + { "type": "MinLength", "config": { "minLength": 3 } }, + { "type": "Print", "config": { "prefix": "[json main] " } } +])json"); + +JsonFilterGraph pipeline(pipelineConfig); +pipeline.filter(std::string{"Hello, filterGraph!"}); +``` + +Output: ```text -/0: unknown filter type 'Uppercas' — did you mean 'Uppercase'? (known types: Join, Reverse, Uppercase, ...) -/1/config/paths/0/0: unknown filter type 'Revrse' — did you mean 'Reverse'? (known types: ...) +[json tap] HELLO, FILTERGRAPH! +[json main] !hparGretlif ,olleH ``` -It catches unknown/typo'd stage types (with a nearest-name suggestion and the -list of known types), structural mistakes (a non-object stage, a missing -`type`, a composite's sub-paths that aren't an array), adjacent type mismatches -between leaf stages, and bad or missing per-stage `config` (leaf stages are -constructed to check). Two things it deliberately does **not** check: a graph's -own declared input/output types (those live in C++ template parameters, not -JSON), and the through-type *across* a composite stage (Fanout/Join), whose -type is branch-/combiner-defined and not knowable from JSON alone. +How the DSL constructs map to JSON: + +| DSL | JSON | +| --- | --- | +| a second reader of an edge, ending in `end` | a `Fanout` stage with `"branches"` | +| several readers of one edge, then `(a, b) -> Merge` | a `Join` stage (`registerJoinFilter`) with `"paths"` and a C++ combiner | +| `Stage(key=value)` | `{ "type": "Stage", "config": { "key": value } }` | +| several outputs (`out.`) | not expressible; a JSON chain has one output | + +`validateGraph(config)` checks a JSON config without running messages and +returns every problem, located by a JSON pointer such as `/1/config/paths/0/0`. +Unlike the DSL checks, it cannot see the graph's declared input/output types, +and it stops type-checking across a `Fanout` or `Join`. ## Build & run this example @@ -287,3 +401,18 @@ cmake --build --preset windows-msvc-release-user-mode On Linux/macOS, use a matching preset such as `unixlike-gcc-release` or `unixlike-clang-release`. + +The complete output: + +```text +[compile-time] !HPARGRETLIF ,OLLEH +[tap] HELLO, FILTERGRAPH! +[main] !hparGretlif ,olleH +[tap] AB +[merge] HELLO, FILTERGRAPH! | !hparGretlif ,olleH (1 hole) +[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 +[json tap] HELLO, FILTERGRAPH! +[json main] !hparGretlif ,olleH +``` diff --git a/README.md b/README.md index e389715..4fba902 100644 --- a/README.md +++ b/README.md @@ -2,13 +2,13 @@ A small, header-only C++20 library for building message/data processing pipelines out of composable filter stages — both at **compile time** (fully -type-safe, zero-overhead composition) and at **runtime** (JSON-configured, -type-erased, with validation at construction time). +type-safe, zero-overhead composition) and at **runtime**, where a graph is +described in a small **text DSL** and type-checked when it is built. It grew out of a need to turn a fixed sequence of transformation steps into a -flexible, reconfigurable **filter graph**: chain stages, fan out to multiple -parallel branches, and drop/short-circuit messages — all without hard-coding -the pipeline shape in source code. +flexible, reconfigurable **filter graph**: chain stages, fan out to parallel +paths, merge them back together, and drop/short-circuit messages — all without +hard-coding the pipeline shape in source code. > **New here? Start with the [example walkthrough (EXAMPLE.md)](EXAMPLE.md)** — > a step-by-step, diagrammed tour of the runnable @@ -24,79 +24,86 @@ the pipeline shape in source code. ## At a glance -A pipeline is a chain of `MessageFilter` stages: each consumes a value and -returns an `std::optional`, where `std::nullopt` short-circuits (drops) the -message. A `FanoutFilter` can duplicate a message to parallel branches while -passing the original through unchanged. +Stages are C++ classes registered under a name. A graph wires them together +with named edges: + +```text +in -> Parse -> msg +msg -> Log -> end # a second reader of msg: gets its own copy +msg -> Validate -> valid -> Format -> out.text +(msg, valid) -> Summarize -> out.stats # fan-in through a merge stage +``` ```mermaid flowchart LR - In(["input"]) --> A["Stage A"] - A --> F{{"FanoutFilter"}} - F -. "copy" .-> B["side branch
(tap)"] - F ==>|"original"| C["Stage C"] - C -->|"optional"| Out(["output"]) - C -. "nullopt" .-> Drop[["dropped"]] + In(["in"]) --> Parse["Parse"] + Parse --> Msg(["msg"]) + Msg -. "copy" .-> Log["Log"] --> End[["end"]] + Msg --> Validate["Validate"] + Validate --> Valid(["valid"]) + Validate -. "nullopt" .-> Drop[["dropped"]] + Valid --> Format["Format"] --> Text(["out.text"]) + Msg --> Summarize{{"Summarize
(merge)"}} + Valid --> Summarize + Summarize --> Stats(["out.stats"]) ``` -See [EXAMPLE.md](EXAMPLE.md) for a full, diagrammed walkthrough of the runnable -[`apps/textPipeline`](apps/textPipeline/main.cpp) sample. +Each stage consumes a value and returns an `std::optional`; `std::nullopt` +drops the message, and everything downstream of that edge is skipped. ## Features +### Stages and compile-time graphs + - **`MessageFilter`** — the base stage interface. A filter consumes `InputType&&` and returns `std::optional`; - returning `std::nullopt` short-circuits (drops) the message, terminating - the chain early. The short-circuit is handled by the framework - (`FilterGraph` / `AnyFilterChain`): once a stage yields `std::nullopt`, no - later stage runs and the whole chain returns `std::nullopt`. Downstream - stages therefore never receive an empty optional — a stage only decides - whether to emit `std::nullopt` itself and never has to handle one as input. - Inside a `FanoutFilter` branch this termination is local to that branch and - does not affect the main path (the branch result is discarded regardless). -- **`Void`** — an explicit terminal marker type. A path normally ends in a - stage that produces a real `OutputType` (the graph's result). Declaring a - stage `MessageFilter` instead marks the path as a pure - side-effect *sink*: it produces no consumable output, distinct from returning - `std::nullopt`, which means a message was *dropped* or could not be processed. - A `Void` stage must be the last stage in a path; `AnyFilterChain` rejects any - stage placed after it at construction time. + returning `std::nullopt` short-circuits (drops) the message. The framework + handles the short-circuit: later stages never receive an empty optional, so a + stage only decides whether to emit `std::nullopt` itself. +- **`Void`** — an explicit terminal marker type. A stage declared + `MessageFilter` is a pure side-effect *sink*: it produces no + consumable output, which is distinct from returning `std::nullopt` (a + *dropped* message). - **`FilterGraph`** — compile-time, variadic-template composition - of stages. Fully type-checked at compile time; each stage's `OutType` must - match the next stage's `InType`. Zero runtime configuration overhead. -- **`AnyMessageFilter`** / **`AnyMessageFilterAdapter`** — type-erased - view of a `MessageFilter`, used to store/chain stages of different - (otherwise incompatible) types at runtime. -- **`FilterRegistry`** / **`FilterRegistrar`** — a global registry - mapping string names to filter factories, so a runtime configuration (e.g. - JSON) can select and construct filters by name. Supports both - parameterless filters and filters configured from a JSON object. -- **`AnyFilterChain`** — builds and validates a sequence of stages ("a path") - from a JSON array, resolving each stage via `FilterRegistry`. Fails fast at - construction time if two consecutive stages' types don't match. -- **`validateGraph(json)` / `Diagnostic`** — pre-flight validation for a config. - It walks the JSON *without running any messages* and returns **all** problems - at once (not just the first), each located by a JSON pointer: unknown/typo'd - stage types (with a nearest-name suggestion), structural mistakes, adjacent - leaf type mismatches, and bad/missing per-stage `config`. Complements the - fail-fast construction-time check with author-friendly, located diagnostics. -- **`JsonFilterGraph`** — a typed wrapper around - `AnyFilterChain`, exposing it as a regular `MessageFilter` so a JSON-configured pipeline can be used anywhere a - compile-time one can. -- **`FanoutFilter`** — duplicates an incoming message across - multiple independent, multi-stage branches (side-effecting "taps": logging, - forwarding, metrics, ...), then passes the *original* message through - unchanged to the rest of the chain. Each branch is itself an - `AnyFilterChain`, so branches can have several stages, not just one. **No - branch contributes to the pipeline's output**: every branch receives its own - copy, runs to completion, and has its result discarded (a branch returning - `std::nullopt` is a no-op for the main path). The value forwarded downstream - is always the unchanged original input, regardless of the number or order of - branches. + of stages. Each stage's `OutType` must match the next stage's `InType`; zero + runtime configuration overhead. - **`SinkFilter`** — a generic terminal stage that forwards data to - a caller-supplied `std::function` callback and returns a simple status - code, useful for terminating a compile-time `FilterGraph`. + a caller-supplied `std::function` callback. + +### Runtime graphs in the text DSL + +- **`FilterRegistry`** / **`FilterRegistrar`** — a global registry + mapping names to stage factories, so a graph description can select and + configure stages by name. +- **`DslFilterGraph`** — builds a graph from DSL text + and exposes it as a regular `MessageFilter`, so it plugs in anywhere a + compile-time graph can (including as a registered stage inside another + graph). Construction instantiates every stage and checks the whole graph up + front: syntax, unknown stage names (with a "did you mean"), bad config, type + mismatches along every edge, fan-in and `Void` misuse, and the output shape. +- **`GraphOutputs`** — the result of a graph with several, optionally named, + outputs of different types (`DslFilterGraph`). +- **`MergeFilter`** / **`registerMergeFilter`** — fan-in stages: + a C++ combiner turns the values of several edges into one, seeing an empty + slot ("hole") for every edge whose path dropped the message. +- **`validateDslGraph(text)`** — the same checks as construction, + returned as a list of `line:column` diagnostics instead of a thrown + **`GraphError`**. +- **`dsl::parseGraphProgram`** / **`dsl::toMermaid`** — parse a graph into its + node/edge form and render it as a Mermaid flowchart. The parser is built on + [lexy](https://github.com/foonathan/lexy); the earlier hand-written parser is + deprecated (`dsl::parseGraphProgramHandwritten` in `GraphLangHandwritten.hpp`). + +### JSON format (still supported) + +- **`JsonFilterGraph`** / **`AnyFilterChain`** — build + a chain from a JSON array of stages. +- **`FanoutFilter`** / **`JoinFilter`** — + the JSON format's composite stages for side branches and scatter-gather. +- **`validateGraph(json)`** — pre-flight validation of a JSON config, located + by JSON pointers. +- **`AnyMessageFilter`** / **`AnyMessageFilterAdapter`** — the + type-erased stage view that both runtime formats are built on. ## Prerequisites @@ -104,13 +111,14 @@ See [EXAMPLE.md](EXAMPLE.md) for a full, diagrammed walkthrough of the runnable `std::format`, ``, and other C++20 features. - **CMake ≥ 3.22** (the presets require CMake ≥ 3.21). - A build generator such as **Ninja** (used by the bundled presets). -- [nlohmann/json](https://github.com/nlohmann/json) (v3.11.3) — fetched +- [nlohmann/json](https://github.com/nlohmann/json) (v3.11.3) and + [lexy](https://github.com/foonathan/lexy) (v2025.05.0) — fetched automatically via CPM; no manual install needed. - Building the tests additionally fetches [Catch2](https://github.com/catchorg/Catch2) (v3.5.2) via CPM. -The library itself is **header-only**: consumers only need a C++20 compiler and -nlohmann/json. +The library itself is **header-only**: consumers only need a C++20 compiler, +nlohmann/json and lexy. ## Installation (CPM) @@ -124,8 +132,9 @@ CPMAddPackage( target_link_libraries(myTarget PRIVATE filterGraph::filterGraph) ``` -`filterGraph` depends on [nlohmann/json](https://github.com/nlohmann/json), -which is pulled in transitively via CPM. +`filterGraph` depends on [nlohmann/json](https://github.com/nlohmann/json) and +[lexy](https://github.com/foonathan/lexy), which are pulled in transitively +via CPM. ## Quick start @@ -154,19 +163,17 @@ FilterGraph pipeline(std::make_shared(), std::make_sha auto result = pipeline.filter(21); // -> "42" ``` -### Runtime, JSON-configured pipeline with a parallel branch +### Runtime pipeline in the DSL, with a tap ```cpp -#include +#include #include -#include #include using namespace filterGraph; -// A side-effecting "tap": logs every value it sees, then passes it on -// unchanged. This is the kind of stage a fanout branch is meant for. +// A side-effecting "tap": logs every value it sees, then passes it on. class Log : public MessageFilter { public: @@ -177,103 +184,183 @@ public: } }; -// Register each stage under a name. FilterRegistrar comes from filterGraph -// (FilterRegistry.hpp); the registration happens in its constructor, so these -// are just static objects — the variable names (registerDouble, ...) are -// arbitrary and never referenced again. The string is the name used in JSON. +// Register each stage under the name the graph uses. The registration happens +// in FilterRegistrar's constructor, so these are just static objects; their +// variable names are arbitrary and never referenced again. static FilterRegistrar registerDouble("Double"); static FilterRegistrar registerToString("ToString"); static FilterRegistrar registerLog("Log"); -static const bool sRegisterFanout = [] { registerFanoutFilter("Fanout"); return true; }(); -// Tap the incoming value into a logging branch, then transform it on the main -// path: double it and turn it into a string. -auto config = nlohmann::json::parse(R"([ - { "type": "Fanout", "config": { "branches": [ - [ { "type": "Log" } ] - ] } }, - { "type": "Double" }, - { "type": "ToString" } -])"); +// Both statements read `in`, so each receives its own copy: the first logs it +// and discards the result at `end`; the second doubles it and turns it into +// the graph's string output. +DslFilterGraph pipeline(R"dsl( + in -> Log -> end + in -> Double -> doubled -> ToString -> out +)dsl"); -JsonFilterGraph pipeline(config); -auto result = pipeline.filter(21); // branch logs "[log] 21"; main path -> "42" +auto result = pipeline.filter(21); // logs "[log] 21"; result -> "42" ``` -The fanout branch observes the *original* input (`21`) as a side effect, while -the main path keeps flowing and produces the transformed result (`"42"`). +If the text had a typo, a config error or a type mismatch, the constructor +would throw a `GraphError` listing every problem with its `line:column`. -**For a complete, diagrammed walkthrough of these concepts, see -[EXAMPLE.md](EXAMPLE.md).** It builds on the runnable -[`apps/textPipeline`](apps/textPipeline/main.cpp) sample (a small text pipeline: -uppercase/reverse/print, with a JSON-configured variant including a fanout -branch and a length-based filter). +**For a complete, diagrammed walkthrough — fan-in merges, named outputs, +diagnostics and the JSON format — see [EXAMPLE.md](EXAMPLE.md).** -## JSON pipeline schema +## The graph language -A pipeline (or a `FanoutFilter` branch) is described as an array of stages: +A graph is a list of statements, one per line. Each statement alternates +**edges** and **stages**, joined by `->`: -```json -[ - { "type": "StageName" }, - { "type": "OtherStage", "config": { "someParam": 123 } } -] +```text +edge -> Stage -> edge -> Stage(key=value) -> edge ``` -- `type` is the name a filter was registered under via `FilterRegistrar`. -- `config` is optional and is passed verbatim to the filter's registered - creator function; its shape is entirely up to that filter. +- **Edges** are names (`msg`, `valid`, ...) that carry one value per message. + Every edge is written by exactly one stage and must be read by something. +- **Stages** are names registered with `FilterRegistrar` (or + `registerMergeFilter`). A stage reads the edge to its left and writes the + edge to its right. +- **Stage arguments** — `Name(key=value, ...)` — are passed to the stage's + registered creator as a JSON object. Values can be numbers, `"strings"`, + `true`/`false`, or barewords (read as strings). A stage without parentheses + receives an empty object. +- **Reserved edge names:** + + | Name | Meaning | + | --- | --- | + | `in` | the graph's input; may only start a statement | + | `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) | + +- **Fan-out:** read the same edge in several statements. Every reader gets its + own copy of the value. +- **Fan-in:** `(a, b, c) -> Merge -> merged` gathers several edges into a merge + stage registered with `registerMergeFilter(name, combiner)`. The + combiner receives `MergeInputs` (`std::vector`), one slot per edge, + in the order listed. +- `#` starts a comment that runs to the end of the line. + +### How a graph runs + +For every message: + +1. Stages run in the order they are written, except that a stage waits until + every edge it reads has been produced. +2. Every reader of an edge gets its own copy (the last reader receives it by + move). +3. A stage returning `std::nullopt` leaves its edge empty; stages reading an + empty edge are skipped, so the drop propagates downstream. +4. A merge receives an empty slot (a "hole") for each dropped edge. It is + skipped only when *all* of its edges are empty. +5. Values routed to `end` are discarded. + +### Choosing the output type + +`DslFilterGraph`'s `OutputType` states what the graph's +outputs must look like, and is checked at construction: + +| `OutputType` | Graph must have | `filter()` returns | +| --- | --- | --- | +| a concrete type, e.g. `std::string` | exactly one `-> out` of that type | the value, or `std::nullopt` if it was dropped | +| `GraphOutputs` (the default) | one or more `-> out` / `-> out.` | all outputs; `std::nullopt` only if every one was dropped | +| `Void` | no outputs, only `-> end` | `Void{}` | -A `FanoutFilter`'s config has one special key, `branches`: an array where -each entry is itself a full stage array (a path), not a single stage: +```cpp +DslFilterGraph analysis(R"dsl( + in -> Uppercase -> out.upper + in -> Length -> out.length +)dsl"); + +auto outputs = analysis.filter(std::string{"hello"}); +outputs->get("upper"); // std::optional{"HELLO"} +outputs->get(1); // outputs can also be read by position +outputs->has("length"); // false if that output's path dropped the message +``` -```json +### Checking and visualizing a graph + +`validateDslGraph` runs every construction check without running a message +and without throwing: + +```cpp +for (const auto& d : validateDslGraph(text)) { - "type": "Fanout", - "config": { - "branches": [ - [ { "type": "StageA" } ], - [ { "type": "StageB" }, { "type": "StageC" } ] - ] - } + std::cerr << dsl::formatDiagnostic(d) << '\n'; } ``` -Branches are side-effect-only: each receives a copy of the message and its -result is discarded, so no branch contributes to the pipeline's output (the -fanout always forwards the unchanged original downstream). +```text +1:7: unknown stage type 'Uppercas' — did you mean 'Uppercase'? +2:7: could not construct 'MinLength': [json.exception.out_of_range.403] key 'minLength' not found +``` -### Validating a config +Syntax errors point at the offending token and say what was expected and what +was found, e.g. `1:17: expected '->' but found 'Print'`. A syntax error ends its +statement, so each line reports at most one and there are no follow-on errors +from half-parsed statements. -Construction throws on the first error with no location, which is awkward while -authoring. `validateGraph` checks a config up front — without running any -messages — and returns *every* problem it finds, each with a JSON pointer: +Type mismatches name the stage, the edge and both types. The type names come +from `typeid(...).name()`, so how they read depends on the compiler. -```cpp -#include +`dsl::toMermaid(dsl::parseGraphProgram(text))` (or +`dsl::toMermaid(graph.program())`) renders a graph as a Mermaid flowchart, +listing each stage's arguments. -auto diagnostics = filterGraph::validateGraph(config); -for (const auto& d : diagnostics) -{ - std::cerr << d.pointer << ": " << d.message << '\n'; -} -if (diagnostics.empty()) { /* safe to build the JsonFilterGraph */ } +### Current limitations + +- 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). + +## JSON format + +The JSON format predates the DSL and remains fully supported; both use the +same registered stages. A pipeline (or a branch/path inside a composite stage) +is an array of stages: + +```json +[ + { "type": "StageName" }, + { "type": "OtherStage", "config": { "someParam": 123 } } +] ``` -Example output for a config with a typo and a nested mistake: +- `type` is the name a filter was registered under via `FilterRegistrar`. +- `config` is optional and is passed verbatim to the filter's registered + creator function. -```text -/0: unknown filter type 'Uppercas' — did you mean 'Uppercase'? (known types: ...) -/1/config/paths/0/0: unknown filter type 'Revrse' — did you mean 'Reverse'? (known types: ...) +JSON chains are linear. Branching is expressed with composite stages whose +config holds nested paths: + +- **`Fanout`** (`registerFanoutFilter`) — `"branches"`: side paths that each + receive a copy and whose results are discarded; the original continues down + the chain. In the DSL this is simply a second reader of an edge ending in + `end`. +- **`Join`** (`registerJoinFilter`) — `"paths"`: scatter one message + through N paths and combine their outputs with a C++ combiner. In the DSL + this is several statements reading the same edge, followed by a merge. + +```cpp +auto config = nlohmann::json::parse(R"([ + { "type": "Fanout", "config": { "branches": [ + [ { "type": "Log" } ] + ] } }, + { "type": "Double" }, + { "type": "ToString" } +])"); + +JsonFilterGraph pipeline(config); // same behaviour as the DSL quick start ``` -It detects unknown/typo'd `type` names (with a nearest-name suggestion and the -list of known types), structural errors (non-object stage, missing `type`, a -composite's sub-paths not being an array), adjacent leaf type mismatches, and -bad/missing per-stage `config` (leaf stages are constructed to check). It does -**not** check a graph's declared input/output types (those are C++ template -parameters, not JSON), and type-chaining pauses across a composite stage -(`Fanout`/`Join`), whose through-type is not knowable from JSON alone. +`validateGraph(config)` checks a JSON config without running messages and +returns every problem, each located by a JSON pointer (e.g. +`/1/config/paths/0/0`). Unlike the DSL checks, it cannot see a graph's declared +input/output types, and type checking pauses across a `Fanout`/`Join`. ## Building & testing @@ -314,14 +401,13 @@ Run the bundled example directly after building: Planned improvements, not yet implemented: -- **Track which stage short-circuited.** When a chain returns `std::nullopt`, - `AnyFilterChain` currently gives no indication of *which* stage dropped the - message. It should record the index/name of the stage that returned - `std::nullopt` (and expose it to the caller, e.g. via an accessor or a - richer result type) so callers can observe and diagnose where a message was - filtered out. (The `Void` type already distinguishes an *intentional* - dead-end from a dropped message; this item covers observing *unintentional* - drops.) +- **Track which stage short-circuited.** When a graph drops a message, neither + `DslFilterGraph` nor `AnyFilterChain` currently tells the caller *which* + stage returned `std::nullopt`. They should record the stage (name and + location) and expose it, e.g. via an accessor or a richer result type, so + callers can diagnose where a message was filtered out. (The `Void` type + already distinguishes an *intentional* dead-end from a dropped message; this + item covers observing *unintentional* drops.) ## License diff --git a/apps/textPipeline/main.cpp b/apps/textPipeline/main.cpp index 7ebb12a..5a07de1 100644 --- a/apps/textPipeline/main.cpp +++ b/apps/textPipeline/main.cpp @@ -2,13 +2,15 @@ // demonstrates the core building blocks of filterGraph: // - MessageFilter: the base stage interface // - FilterGraph: compile-time chaining of stages -// - FanoutFilter + FilterRegistry + JsonFilterGraph: a runtime, JSON -// configured pipeline with a parallel branch ("tap") -// - JoinFilter: scatter-gather, the mirror of Fanout (N paths -> 1 output) +// - FilterRegistry + DslFilterGraph: runtime graphs described in the text DSL, +// with fan-out (a "tap"), fan-in (a merge) and several named outputs +// - validateDslGraph: every problem in a broken graph, located by line:column +// - JsonFilterGraph: the JSON format, which remains supported +#include #include #include -#include #include +#include #include #include @@ -16,20 +18,24 @@ #include #include #include +#include #include #include #include #include #include -#include -using filterGraph::FanoutFilter; +using filterGraph::DslFilterGraph; using filterGraph::FilterGraph; using filterGraph::FilterRegistrar; using filterGraph::JsonFilterGraph; +using filterGraph::MergeInputs; using filterGraph::MessageFilter; using filterGraph::registerFanoutFilter; -using filterGraph::registerJoinFilter; +using filterGraph::registerMergeFilter; +using filterGraph::validateDslGraph; + +namespace dsl = filterGraph::dsl; // --- Stages ----------------------------------------------------------- @@ -53,9 +59,19 @@ class ReverseFilter : public MessageFilter } }; +// Changes the message type: string in, length out. +class LengthFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + return text.size(); + } +}; + // A configurable stage: drops (short-circuits) any message shorter than a -// configurable minimum length, demonstrating both JSON-driven configuration -// and early chain termination via std::nullopt. +// configurable minimum length, demonstrating both configuration and early +// termination via std::nullopt. class MinLengthFilter : public MessageFilter { public: @@ -95,11 +111,13 @@ class PrintFilter : public MessageFilter std::string mPrefix; }; -// --- Registration for the JSON-driven part of the example -------------- +// --- Registration: the names the DSL (and JSON) refer to --------------- static FilterRegistrar registerUppercase("Uppercase"); -static FilterRegistrar registerReverse("Reverse"); +static FilterRegistrar registerReverse("Reverse"); +static FilterRegistrar registerLength("Length"); +// Stage arguments, e.g. `MinLength(minLength=3)`, arrive as a JSON object. static FilterRegistrar registerMinLength( "MinLength", [](const nlohmann::json& config) { @@ -112,24 +130,18 @@ static FilterRegistrar registerPrint( return std::make_shared(config.value("prefix", std::string{})); }); -static const bool sRegisterFanout = [] { - registerFanoutFilter("Fanout"); - return true; -}(); - -// The Join combiner lives in C++ (it cannot be expressed in JSON): it gathers -// one slot per path (empty slots are "holes" left by paths that dropped their -// message via std::nullopt), concatenates the present outputs with " | ", and -// reports how many holes it saw. -static const bool sRegisterJoin = [] { - registerJoinFilter( - "Join", - [](std::vector&& outputs) -> std::optional { +// A merge stage for fan-in `(a, b, c) -> Concat`. The combiner receives one +// slot per edge; empty slots are "holes" left by paths that dropped the +// message. It concatenates the present values with " | " and counts the holes. +static const bool sRegisterConcat = [] { + registerMergeFilter( + "Concat", + [](MergeInputs&& inputs) -> std::optional { std::string joined; std::size_t holes = 0; - for (auto& out : outputs) + for (auto& input : inputs) { - if (!out.has_value()) + if (!input.has_value()) { ++holes; continue; @@ -138,13 +150,19 @@ static const bool sRegisterJoin = [] { { joined += " | "; } - joined += std::any_cast(out); + joined += std::any_cast(input); } return joined + std::format(" ({} hole{})", holes, holes == 1 ? "" : "s"); }); return true; }(); +// Only needed by the JSON example: in the DSL, fan-out is built in. +static const bool sRegisterFanout = [] { + registerFanoutFilter("Fanout"); + return true; +}(); + int main() { // 1) Compile-time pipeline: FilterGraph @@ -158,42 +176,77 @@ int main() pipeline.filter(std::string{"Hello, filterGraph!"}); } - // 2) Runtime, JSON-configured pipeline with a parallel branch ("tap"): - // the original text is duplicated to a side branch (uppercase+print) - // while the main path reverses it, drops it if too short, and prints. + // 2) Runtime graph in the text DSL, with a tap. Both statements read `in`, + // so each gets its own copy: the first uppercases and prints its copy + // (then discards it at `end`); the second reverses the message, drops it + // if it is too short, and prints it as the graph's output. { - auto pipelineConfig = nlohmann::json::parse(R"([ - { "type": "Fanout", "config": { "branches": [ - [ { "type": "Uppercase" }, { "type": "Print", "config": { "prefix": "[branch] " } } ] - ] } }, - { "type": "Reverse" }, - { "type": "MinLength", "config": { "minLength": 3 } }, - { "type": "Print", "config": { "prefix": "[main] " } } - ])"); + DslFilterGraph pipeline(R"dsl( + in -> Uppercase -> shouted -> Print(prefix="[tap] ") -> end + in -> Reverse -> reversed -> MinLength(minLength=3) -> long -> Print(prefix="[main] ") -> out + )dsl"); + + pipeline.filter(std::string{"Hello, filterGraph!"}); + pipeline.filter(std::string{"ab"}); // tapped, then dropped by MinLength on the main path + } + + // 3) Fan-in: three paths read the same input and a merge combines them. + // MinLength=100 drops its copy, which leaves a hole in the merge. + { + DslFilterGraph pipeline(R"dsl( + in -> Uppercase -> upper + in -> Reverse -> reversed + in -> MinLength(minLength=100) -> long + (upper, reversed, long) -> Concat -> joined -> Print(prefix="[merge] ") -> out + )dsl"); - JsonFilterGraph pipeline(pipelineConfig); pipeline.filter(std::string{"Hello, filterGraph!"}); - pipeline.filter(std::string{"ab"}); // dropped by MinLength on the main path } - // 3) Join (scatter-gather): the mirror of Fanout. The SAME input is - // scattered (copied) through N independent paths; a C++ combiner then - // gathers their outputs into one. Here two paths transform the text - // (uppercase, reverse) and a third drops it (MinLength=100), leaving a - // hole the combiner can see and report. + // 4) Several named outputs of different types, returned as GraphOutputs. + { + DslFilterGraph analysis(R"dsl( + in -> Uppercase -> out.upper + in -> Length -> out.length + in -> MinLength(minLength=100) -> out.long + )dsl"); + + auto outputs = analysis.filter(std::string{"Hello, filterGraph!"}); + std::cout << std::format("[outputs] upper={} length={} long={}\n", + *outputs->get("upper"), + *outputs->get("length"), + outputs->has("long") ? "present" : "dropped"); + } + + // 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. { - auto joinConfig = nlohmann::json::parse(R"([ - { "type": "Join", "config": { "paths": [ - [ { "type": "Uppercase" } ], - [ { "type": "Reverse" } ], - [ { "type": "MinLength", "config": { "minLength": 100 } } ] + const auto diagnostics = validateDslGraph( + "in -> Uppercas -> shouted -> Print -> out\n" + "in -> MinLength -> long -> Print -> end\n"); + + for (const auto& diagnostic : diagnostics) + { + std::cout << "[check] " << dsl::formatDiagnostic(diagnostic) << '\n'; + } + } + + // 6) The JSON format is still supported and uses the same registry. This is + // the tap pipeline from (2), written with a Fanout stage. + { + auto pipelineConfig = nlohmann::json::parse(R"json([ + { "type": "Fanout", "config": { "branches": [ + [ { "type": "Uppercase" }, { "type": "Print", "config": { "prefix": "[json tap] " } } ] ] } }, - { "type": "Print", "config": { "prefix": "[join] " } } - ])"); + { "type": "Reverse" }, + { "type": "MinLength", "config": { "minLength": 3 } }, + { "type": "Print", "config": { "prefix": "[json main] " } } + ])json"); - JsonFilterGraph pipeline(joinConfig); + JsonFilterGraph pipeline(pipelineConfig); pipeline.filter(std::string{"Hello, filterGraph!"}); } return 0; -} \ No newline at end of file +} diff --git a/libs/filterGraph/CMakeLists.txt b/libs/filterGraph/CMakeLists.txt index 8ccb0e0..7f6d395 100644 --- a/libs/filterGraph/CMakeLists.txt +++ b/libs/filterGraph/CMakeLists.txt @@ -13,6 +13,13 @@ target_sources(${PROJECT_NAME} core/filterGraph/AnyFilterChain.hpp core/filterGraph/JsonFilterGraph.hpp core/filterGraph/FanoutFilter.hpp + core/filterGraph/JoinFilter.hpp + core/filterGraph/MergeFilter.hpp + core/filterGraph/GraphValidator.hpp + core/filterGraph/GraphLang.hpp + core/filterGraph/GraphLangLexy.hpp + core/filterGraph/GraphLangHandwritten.hpp + core/filterGraph/DslFilterGraph.hpp ) target_link_libraries(${PROJECT_NAME} diff --git a/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp new file mode 100644 index 0000000..8fb497c --- /dev/null +++ b/libs/filterGraph/core/filterGraph/DslFilterGraph.hpp @@ -0,0 +1,603 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// DslFilterGraph runs a filter graph described in the text DSL (syntax: see +// GraphLang.hpp), e.g. +// +// in -> Uppercase -> shouted -> Print(prefix="[tap] ") -> end +// in -> Reverse -> reversed -> MinLength(minLength=3) -> long -> Print -> out +// +// Execution semantics, per message: +// - stages run in source order, except that a stage waits until every edge it +// reads has been produced; +// - 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 +// 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; +// - values routed to `end` are discarded; a stage producing Void must route +// to `end`. +namespace filterGraph { + +// The results of a graph with several outputs: one slot per `-> out` / +// `-> out.`, in DSL order. An empty slot means that output's path dropped +// the message. +class GraphOutputs +{ +public: + using Keys = std::vector>; + + GraphOutputs(std::vector slots, std::shared_ptr keys) + : mSlots(std::move(slots)) + , mKeys(std::move(keys)) + { + } + + std::size_t size() const + { + return mSlots.size(); + } + + const Keys& keys() const + { + return *mKeys; + } + + bool has(std::size_t index) const + { + return mSlots.at(index).has_value(); + } + + bool has(std::string_view key) const + { + return has(indexOf(key)); + } + + // Returns a copy of the output, or std::nullopt if its path dropped the + // message. Throws std::bad_any_cast if T is not the output's type. + template + std::optional get(std::size_t index) const + { + const std::any& slot = mSlots.at(index); + if (!slot.has_value()) + { + return std::nullopt; + } + return std::any_cast(slot); + } + + template + std::optional get(std::string_view key) const + { + return get(indexOf(key)); + } + + // Like get(), but moves the value out and leaves the slot empty. + template + std::optional take(std::size_t index) + { + std::any& slot = mSlots.at(index); + if (!slot.has_value()) + { + return std::nullopt; + } + std::optional value = std::any_cast(std::move(slot)); + slot.reset(); + return value; + } + + template + std::optional take(std::string_view key) + { + return take(indexOf(key)); + } + +private: + std::size_t indexOf(std::string_view key) const + { + for (std::size_t i = 0; i < mKeys->size(); ++i) + { + if ((*mKeys)[i] && *(*mKeys)[i] == key) + { + return i; + } + } + throw std::out_of_range(std::format("GraphOutputs: the graph has no output named '{}'", key)); + } + + std::vector mSlots; + std::shared_ptr mKeys; +}; + +// 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 +{ +public: + explicit GraphError(std::vector diagnostics) + : std::runtime_error("DslFilterGraph: invalid graph\n" + dsl::formatDiagnostics(diagnostics)) + , mDiagnostics(std::move(diagnostics)) + { + } + + const std::vector& diagnostics() const noexcept + { + return mDiagnostics; + } + +private: + std::vector mDiagnostics; +}; + +namespace dsl::detail { + +// What a typed graph expects at its output boundary. +enum class OutputShape +{ + Single, // exactly one `-> out` of the graph's OutputType + Multiple, // one or more outputs, returned as GraphOutputs + None // no outputs at all: a pure sink graph (OutputType = Void) +}; + +struct ExpectedOutputs +{ + OutputShape shape; + std::type_index type; +}; + +template +ExpectedOutputs expectedOutputs() +{ + if (std::is_same_v) + { + return {OutputShape::Multiple, typeid(GraphOutputs)}; + } + if (std::is_same_v) + { + return {OutputShape::None, typeid(Void)}; + } + return {OutputShape::Single, typeid(OutputType)}; +} + +// Stage indices in run order: source order, except that a stage waits until all +// of its inputs have been produced. Stages that can never run (an input is +// never produced, or a cycle) are left out. +inline std::vector runOrder(const GraphProgram& program) +{ + std::unordered_set produced{"in"}; + std::vector scheduled(program.stages.size(), false); + std::vector order; + + bool progress = true; + while (progress) + { + progress = false; + for (std::size_t i = 0; i < program.stages.size(); ++i) + { + const StageNode& node = program.stages[i]; + const bool ready = std::all_of(node.inputs.begin(), node.inputs.end(), [&](const std::string& edge) { + return produced.contains(edge); + }); + if (scheduled[i] || !ready) + { + continue; + } + scheduled[i] = true; + order.push_back(i); + if (node.output) + { + produced.insert(*node.output); + } + progress = true; + break; // rescan from the top so earlier statements keep priority + } + } + return order; +} + +inline std::string describeEdge(const std::string& edge) +{ + if (edge == "in") + { + return "the graph input 'in'"; + } + return std::format("edge '{}'", edge); +} + +struct CompiledStage +{ + std::shared_ptr filter; + std::vector inputs; // edge slots read by the stage + std::optional output; // edge slot written; nullopt => `end` + bool merge = false; // inputs are gathered into MergeInputs +}; + +// The executable form of a GraphProgram: every stage instantiated through the +// FilterRegistry, type-checked, and wired to numbered edge slots in run order. +class GraphPlan +{ +public: + GraphPlan() = default; + + // Builds the plan, appending every problem found (bad config, type + // mismatches along edges, misused merges/Void, output shape) to + // `diagnostics`. The plan may only be run if no diagnostics were added. + GraphPlan(const GraphProgram& program, + std::type_index inputType, + const ExpectedOutputs& expected, + 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) + + if (program.stages.empty()) + { + diagnostics.push_back({SourceLoc{}, "the graph has no stages; add a statement such as 'in -> Stage -> out'"}); + } + + const std::vector order = runOrder(program); + if (order.size() != program.stages.size() && program.ok()) + { + 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); + + for (std::size_t index : order) + { + const StageNode& node = program.stages[index]; + if (!registry.contains(node.type)) + { + continue; // already reported by parseGraphProgram; its output stays untyped + } + + std::shared_ptr filter; + try + { + filter = registry.create(node.type, node.config); + } + catch (const std::exception& error) + { + diagnostics.push_back({node.loc, std::format("could not construct '{}': {}", node.type, error.what())}); + continue; + } + + const bool merge = filter->inputType() == std::type_index(typeid(MergeInputs)); + if (node.fanIn && !merge) + { + diagnostics.push_back( + {node.loc, + std::format("stage '{}' takes a single input and cannot follow a fan-in group; combine the " + "edges with a merge stage (see registerMergeFilter)", + node.type)}); + } + else if (!node.fanIn && merge) + { + diagnostics.push_back( + {node.loc, + std::format("'{}' is a merge stage; feed it a fan-in group, e.g. '(a, b) -> {} -> merged'", + node.type, + node.type)}); + } + else if (!merge) + { + const std::string& edge = node.inputs.front(); + auto type = edgeTypes.find(edge); + if (type != edgeTypes.end() && type->second != 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())}); + } + } + + if (node.output) + { + if (filter->outputType() == std::type_index(typeid(Void))) + { + diagnostics.push_back( + {node.loc, + std::format("stage '{}' produces Void (no value), so it must route to 'end'", node.type)}); + } + else + { + edgeTypes.emplace(*node.output, filter->outputType()); + } + } + + CompiledStage stage; + stage.filter = std::move(filter); + stage.merge = merge; + for (const auto& edge : node.inputs) + { + stage.inputs.push_back(slotOf(edge)); + } + if (node.output) + { + stage.output = slotOf(*node.output); + } + mStages.push_back(std::move(stage)); + } + + auto keys = std::make_shared(); + for (const auto& binding : program.outputs) + { + mOutputEdges.push_back(slotOf(binding.edge)); + keys->push_back(binding.key); + } + mOutputKeys = std::move(keys); + + checkOutputs(program, expected, edgeTypes, diagnostics); + + mEdgeCount = slots.size(); + mReaders.assign(mEdgeCount, 0); + for (const auto& stage : mStages) + { + for (std::size_t edge : stage.inputs) + { + ++mReaders[edge]; + } + } + } + + // 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 + { + std::vector edges(mEdgeCount); + std::vector remaining = mReaders; + edges[kInputSlot] = std::move(input); + + for (const auto& stage : mStages) + { + std::optional result; + if (stage.merge) + { + MergeInputs gathered; + gathered.reserve(stage.inputs.size()); + for (std::size_t edge : stage.inputs) + { + gathered.push_back(read(edges, remaining, edge)); + } + const bool anyPresent = std::any_of(gathered.begin(), gathered.end(), [](const std::any& slot) { + return slot.has_value(); + }); + if (!anyPresent) + { + continue; // every path into the merge dropped the message + } + result = stage.filter->filter(std::any(std::move(gathered))); + } + else + { + std::any value = read(edges, remaining, stage.inputs.front()); + if (!value.has_value()) + { + continue; // dropped upstream + } + result = stage.filter->filter(std::move(value)); + } + + if (result && stage.output) + { + edges[*stage.output] = std::move(*result); + } + } + + std::vector outputs; + outputs.reserve(mOutputEdges.size()); + for (std::size_t edge : mOutputEdges) + { + outputs.push_back(std::move(edges[edge])); + } + return outputs; + } + + const std::shared_ptr& outputKeys() const + { + return mOutputKeys; + } + +private: + static constexpr std::size_t kInputSlot = 0; + + // The last reader of an edge takes the value by move; earlier readers + // (fan-out) each get their own copy. + static std::any read(std::vector& edges, std::vector& remaining, std::size_t edge) + { + if (--remaining[edge] == 0) + { + return std::move(edges[edge]); + } + return edges[edge]; + } + + static void checkOutputs(const GraphProgram& program, + const ExpectedOutputs& expected, + const std::unordered_map& edgeTypes, + std::vector& diagnostics) + { + const auto& outputs = program.outputs; + switch (expected.shape) + { + case OutputShape::Single: + if (outputs.empty()) + { + diagnostics.push_back( + {SourceLoc{}, + std::format("the graph's output type '{}' needs one '-> out', but the graph has none " + "(use Void as the output type for a graph without outputs)", + expected.type.name())}); + } + else if (outputs.size() > 1) + { + diagnostics.push_back( + {outputs[1].loc, + std::format("the graph's output type '{}' needs exactly one '-> out', but the graph has {} " + "(use GraphOutputs as the output type to receive several)", + expected.type.name(), + outputs.size())}); + } + else if (auto type = edgeTypes.find(outputs[0].edge); + type != edgeTypes.end() && type->second != 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())}); + } + break; + case OutputShape::Multiple: + if (outputs.empty()) + { + diagnostics.push_back({SourceLoc{}, + "GraphOutputs needs at least one '-> out' (use Void as the output type for a " + "graph without outputs)"}); + } + break; + case OutputShape::None: + if (!outputs.empty()) + { + diagnostics.push_back({outputs[0].loc, + "the graph's output type is Void, but the graph routes a value to 'out'; " + "route it to 'end' instead"}); + } + break; + } + } + + std::vector mStages; + std::size_t mEdgeCount = 0; + std::vector mReaders; // per edge slot: number of stage inputs reading it + std::vector mOutputEdges; + std::shared_ptr mOutputKeys; +}; + +} // namespace dsl::detail + +// DslFilterGraph builds a graph from DSL text (or an +// already parsed GraphProgram) and exposes it as a MessageFilter, so it plugs +// in anywhere a compile-time FilterGraph or a JsonFilterGraph can. +// +// 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; +// - GraphOutputs: one or more (optionally keyed) outputs of any types; +// std::nullopt only when every output was dropped; +// - Void: no outputs, only `-> end` (a pure sink graph). +// +// Construction instantiates every stage and checks the whole graph up front: +// syntax, unknown stages, bad config, type mismatches along every edge, merges +// and Void used correctly, and the output shape. All problems are thrown +// together in one GraphError; use validateDslGraph to get them without an +// exception. +template +class DslFilterGraph : public MessageFilter +{ +public: + explicit DslFilterGraph(std::string_view text) + : DslFilterGraph(dsl::parseGraphProgram(text)) + { + } + + explicit DslFilterGraph(dsl::GraphProgram program) + : mProgram(std::move(program)) + { + std::vector diagnostics = mProgram.diagnostics; + mPlan = dsl::detail::GraphPlan( + mProgram, typeid(InputType), dsl::detail::expectedOutputs(), diagnostics); + if (!diagnostics.empty()) + { + dsl::detail::sortDiagnostics(diagnostics); + throw GraphError(std::move(diagnostics)); + } + } + + std::optional filter(InputType&& data) override + { + std::vector outputs = mPlan.run(std::any(std::move(data))); + + if constexpr (std::is_same_v) + { + return Void{}; + } + else if constexpr (std::is_same_v) + { + const bool anyPresent = std::any_of(outputs.begin(), outputs.end(), [](const std::any& slot) { + return slot.has_value(); + }); + if (!anyPresent) + { + return std::nullopt; + } + return GraphOutputs(std::move(outputs), mPlan.outputKeys()); + } + else + { + if (!outputs.front().has_value()) + { + return std::nullopt; + } + return std::any_cast(std::move(outputs.front())); + } + } + + // The parsed graph, e.g. for dsl::toMermaid. + const dsl::GraphProgram& program() const + { + return mProgram; + } + +private: + dsl::GraphProgram mProgram; + dsl::detail::GraphPlan mPlan; +}; + +// Checks DSL text for use as DslFilterGraph without +// running any messages, and returns every problem found (sorted by location) +// instead of throwing. Stages are instantiated to check their config and types. +template +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); + dsl::detail::sortDiagnostics(diagnostics); + return diagnostics; +} + +} // namespace filterGraph diff --git a/libs/filterGraph/core/filterGraph/GraphLang.hpp b/libs/filterGraph/core/filterGraph/GraphLang.hpp index 646b848..4bc1da6 100644 --- a/libs/filterGraph/core/filterGraph/GraphLang.hpp +++ b/libs/filterGraph/core/filterGraph/GraphLang.hpp @@ -3,6 +3,11 @@ #include #include +#include +#include +#include +#include + #include #include @@ -10,32 +15,41 @@ #include #include #include +#include #include #include #include #include +#include #include -// Text DSL (Model B) for describing a filter graph as a named-edge DAG. +// Text DSL for describing a filter graph as a named-edge DAG. This is the +// preferred way to describe a runtime graph; DslFilterGraph (DslFilterGraph.hpp) +// builds and runs one. // // A program is a set of newline-separated statements; each statement is an // alternating chain of edges and stages joined by "->": // // in -> Parse -> msg // msg -> Format -> out.text -// msg -> Summary -> out.stats -// (a, b) -> Merge -> c +// msg -> Summary -> sum +// (msg, sum) -> Merge -> out.stats // msg -> Store -> end // // - even positions are edges (identifiers) or the reserved boundary nodes -// `in` (graph input), `out` / `out.` (results), `end` (dead-end/Void); +// `in` (graph input), `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; -// - fan-in = a source group `(a, b) -> Merge -> c`. +// - fan-in = a source group `(a, b) -> Merge -> c`, where Merge is a merge +// stage (see registerMergeFilter in MergeFilter.hpp); +// - `#` starts a comment that runs to the end of the line. // -// This front-end only parses/validates/visualizes; executing a Model-B graph -// needs runtime edge-routing that is intentionally out of scope here. +// This header parses (with a lexy-based parser) and structurally validates a +// program into a node/edge IR (GraphProgram) and renders it with toMermaid; +// DslFilterGraph instantiates, type-checks and runs it. The original +// hand-written parser is deprecated (GraphLangHandwritten.hpp). namespace filterGraph::dsl { struct SourceLoc @@ -57,7 +71,8 @@ struct StageNode std::string type; // registered filter/merge name nlohmann::json config; // parsed config args std::vector inputs; // source edge names ("in" allowed) - std::optional output; // produced edge name; nullopt => routed to out/end + std::optional output; // produced edge name ("$outN" for `out`); nullopt => routed to end + bool fanIn = false; // inputs came from a group `(a, b) -> Stage` SourceLoc loc; }; @@ -83,36 +98,47 @@ struct GraphProgram } }; -namespace detail { +// Formats a diagnostic as "line:column: message". +inline std::string formatDiagnostic(const TextDiagnostic& diagnostic) +{ + return std::format("{}:{}: {}", diagnostic.loc.line, diagnostic.loc.column, diagnostic.message); +} -inline bool isReserved(std::string_view name) +// Formats diagnostics one per line. +inline std::string formatDiagnostics(const std::vector& diagnostics) { - return name == "in" || name == "out" || name == "end"; + std::string text; + for (const auto& diagnostic : diagnostics) + { + if (!text.empty()) + { + text += '\n'; + } + text += formatDiagnostic(diagnostic); + } + return text; } -// ------------------------------- tokenizer ------------------------------- +namespace detail { -enum class TokenKind +inline void sortDiagnostics(std::vector& diagnostics) { - Identifier, - Number, - String, - Arrow, // -> - LParen, // ( - RParen, // ) - Comma, // , - Equals, // = - Dot, // . - Newline, - EndOfInput -}; + std::stable_sort(diagnostics.begin(), diagnostics.end(), [](const TextDiagnostic& a, const TextDiagnostic& b) { + return a.loc.line != b.loc.line ? a.loc.line < b.loc.line : a.loc.column < b.loc.column; + }); +} -struct Token +inline bool isReserved(std::string_view name) { - TokenKind kind; - std::string text; - SourceLoc loc; -}; + return name == "in" || name == "out" || name == "end"; +} + +// ------------------------- shared lexical helpers ------------------------ +// +// The parser below and the deprecated hand-written parser +// (GraphLangHandwritten.hpp) classify input and produce literal values and +// error descriptions through these helpers, so they report identical +// diagnostics. inline bool isIdentStart(char c) { @@ -129,193 +155,141 @@ inline bool isDigit(char c) return c >= '0' && c <= '9'; } -class Lexer +struct StringLiteral { -public: - explicit Lexer(std::string_view source) - : mSource(source) - { - } + std::string value; // with escapes resolved + std::size_t length; // in the source, including both quotes +}; - std::vector tokenize() +// `rest` starts at an opening quote. A backslash takes the next character +// literally. Returns std::nullopt if the line ends before the closing quote. +inline std::optional scanStringLiteral(std::string_view rest) +{ + std::string value; + std::size_t i = 1; + while (i < rest.size() && rest[i] != '"' && rest[i] != '\n') { - std::vector tokens; - while (mPos < mSource.size()) + if (rest[i] == '\\' && i + 1 < rest.size() && rest[i + 1] != '\n') { - const char c = mSource[mPos]; - - if (c == '\n') - { - tokens.push_back(make(TokenKind::Newline, "\n")); - advance(); - continue; - } - if (c == '\r' || c == ' ' || c == '\t') - { - advance(); - continue; - } - if (c == '#') - { - while (mPos < mSource.size() && mSource[mPos] != '\n') - { - advance(); - } - continue; - } - if (c == '-' && mPos + 1 < mSource.size() && mSource[mPos + 1] == '>') - { - tokens.push_back(make(TokenKind::Arrow, "->")); - advance(); - advance(); - continue; - } - if (c == '(') - { - tokens.push_back(make(TokenKind::LParen, "(")); - advance(); - continue; - } - if (c == ')') - { - tokens.push_back(make(TokenKind::RParen, ")")); - advance(); - continue; - } - if (c == ',') - { - tokens.push_back(make(TokenKind::Comma, ",")); - advance(); - continue; - } - if (c == '=') - { - tokens.push_back(make(TokenKind::Equals, "=")); - advance(); - continue; - } - if (c == '.') - { - tokens.push_back(make(TokenKind::Dot, ".")); - advance(); - continue; - } - if (c == '"') - { - tokens.push_back(lexString()); - continue; - } - if (isDigit(c) || (c == '-' && mPos + 1 < mSource.size() && isDigit(mSource[mPos + 1]))) - { - tokens.push_back(lexNumber()); - continue; - } - if (isIdentStart(c)) - { - tokens.push_back(lexIdentifier()); - continue; - } - - // Unknown character: record and skip so lexing can continue. - mErrors.push_back({currentLoc(), std::format("unexpected character '{}'", c)}); - advance(); + ++i; } - tokens.push_back(make(TokenKind::EndOfInput, "")); - return tokens; + value += rest[i]; + ++i; } - - const std::vector& errors() const + if (i >= rest.size() || rest[i] != '"') { - return mErrors; + return std::nullopt; } + return StringLiteral{std::move(value), i + 1}; +} -private: - Token make(TokenKind kind, std::string text) const +// Length of the number token at the start of `rest` (an optional '-', a digit, +// then digits and dots), or 0 if `rest` does not start with a number. +inline std::size_t numberLength(std::string_view rest) +{ + std::size_t i = 0; + if (!rest.empty() && rest[0] == '-') { - return Token{kind, std::move(text), currentLoc()}; + i = 1; } - - SourceLoc currentLoc() const + if (i >= rest.size() || !isDigit(rest[i])) + { + return 0; + } + while (i < rest.size() && (isDigit(rest[i]) || rest[i] == '.')) { - return SourceLoc{mLine, mColumn}; + ++i; } + return i; +} - void advance() +// Converts a number token to JSON: an integer, or a double if it has a +// fractional part. On failure returns std::nullopt and sets `error`. +inline std::optional numberValue(std::string_view text, std::string& error) +{ + const std::string number(text); + const std::size_t dot = number.find('.'); + if (dot != std::string::npos && (dot + 1 == number.size() || number.find('.', dot + 1) != std::string::npos)) { - if (mSource[mPos] == '\n') - { - ++mLine; - mColumn = 1; - } - else + error = std::format("malformed number '{}'", number); + return std::optional{}; + } + try + { + if (dot != std::string::npos) { - ++mColumn; + return nlohmann::json(std::stod(number)); } - ++mPos; + return nlohmann::json(static_cast(std::stoll(number))); } + catch (const std::out_of_range&) + { + error = std::format("number '{}' is out of range", number); + return std::optional{}; + } +} - Token lexString() +inline std::string unexpectedCharacter(char c) +{ + const auto code = static_cast(c); + if (code >= 0x20 && code < 0x7F) { - const SourceLoc start = currentLoc(); - advance(); // opening quote - std::string value; - while (mPos < mSource.size() && mSource[mPos] != '"') - { - if (mSource[mPos] == '\\' && mPos + 1 < mSource.size()) - { - advance(); - } - value += mSource[mPos]; - advance(); - } - if (mPos < mSource.size()) - { - advance(); // closing quote - } - else - { - mErrors.push_back({start, "unterminated string literal"}); - } - return Token{TokenKind::String, std::move(value), start}; + return std::format("unexpected character '{}'", c); } + return std::format("unexpected byte 0x{:02X}", static_cast(code)); +} - Token lexNumber() +// Describes the input at the start of `rest` for an "expected ... but found +// ..." message. If that input is not a valid token at all, `invalid` is set and +// `text` is the complete error message instead. +struct Found +{ + bool invalid = false; + std::string text; +}; + +inline Found describeFound(std::string_view rest) +{ + rest = rest.substr(0, rest.find('\n')); + if (rest.empty() || rest.front() == '#') { - const SourceLoc start = currentLoc(); - std::string value; - if (mSource[mPos] == '-') - { - value += '-'; - advance(); - } - while (mPos < mSource.size() && (isDigit(mSource[mPos]) || mSource[mPos] == '.')) + return {false, "end of line"}; + } + if (rest.starts_with("->")) + { + return {false, "'->'"}; + } + + const char c = rest.front(); + if (c == '"') + { + if (scanStringLiteral(rest)) { - value += mSource[mPos]; - advance(); + return {false, "a string literal"}; } - return Token{TokenKind::Number, std::move(value), start}; + return {true, "unterminated string literal"}; } - - Token lexIdentifier() + if (const std::size_t length = numberLength(rest)) { - const SourceLoc start = currentLoc(); - std::string value; - while (mPos < mSource.size() && isIdentPart(mSource[mPos])) + return {false, std::format("'{}'", rest.substr(0, length))}; + } + if (isIdentStart(c)) + { + std::size_t length = 1; + while (length < rest.size() && isIdentPart(rest[length])) { - value += mSource[mPos]; - advance(); + ++length; } - return Token{TokenKind::Identifier, std::move(value), start}; + return {false, std::format("'{}'", rest.substr(0, length))}; } + if (c == '(' || c == ')' || c == ',' || c == '=' || c == '.') + { + return {false, std::format("'{}'", c)}; + } + return {true, unexpectedCharacter(c)}; +} - std::string_view mSource; - std::size_t mPos = 0; - std::size_t mLine = 1; - std::size_t mColumn = 1; - std::vector mErrors; -}; - -// -------------------------------- parser --------------------------------- +// -------------------------------- terms ---------------------------------- // A single term in a statement, already classified by the parser. struct Term @@ -338,7 +312,7 @@ struct Term // 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 front-end (hand-written or lexy). +// next edge/boundary. Shared by every parser. inline void buildStatement(std::vector& terms, GraphProgram& program, std::size_t& outputIndex, @@ -392,6 +366,11 @@ inline void buildStatement(std::vector& terms, diags.push_back({terms[i].loc, "a fan-in group may only appear as the first term of a statement"}); return; } + if (terms[i].kind == Term::Kind::Group && terms[i].edges.empty()) + { + diags.push_back({terms[i].loc, "a fan-in group needs at least one edge"}); + return; + } } } @@ -400,14 +379,16 @@ inline void buildStatement(std::vector& terms, for (std::size_t i = 1; i < terms.size(); i += 2) { StageNode node; - node.type = terms[i].name; - node.config = terms[i].config; + node.type = terms[i].name; + // A stage without `(...)` gets an empty object, as a JSON stage without "config" does. + node.config = terms[i].config.is_null() ? nlohmann::json::object() : terms[i].config; node.loc = terms[i].loc; const Term& source = terms[i - 1]; if (source.kind == Term::Kind::Group) { node.inputs = source.edges; + node.fanIn = true; } else { @@ -437,270 +418,6 @@ inline void buildStatement(std::vector& terms, } } -class Parser -{ -public: - Parser(std::vector tokens, std::vector lexErrors) - : mTokens(std::move(tokens)) - , mDiagnostics(std::move(lexErrors)) - { - } - - GraphProgram parse() - { - GraphProgram program; - std::size_t outputIndex = 0; - - while (peek().kind != TokenKind::EndOfInput) - { - if (peek().kind == TokenKind::Newline) - { - next(); - continue; - } - - auto terms = parseStatement(); - if (!terms.empty()) - { - buildStatement(terms, program, outputIndex, mDiagnostics); - } - } - - program.diagnostics = std::move(mDiagnostics); - return program; - } - -private: - const Token& peek() const - { - return mTokens[mPos]; - } - - const Token& next() - { - return mTokens[mPos < mTokens.size() - 1 ? mPos++ : mPos]; - } - - void error(const SourceLoc& loc, std::string message) - { - mDiagnostics.push_back({loc, std::move(message)}); - } - - // Parses a single line into a flat list of terms (until newline / EOF). - std::vector parseStatement() - { - std::vector terms; - bool expectArrow = false; - - while (peek().kind != TokenKind::Newline && peek().kind != TokenKind::EndOfInput) - { - if (expectArrow) - { - if (peek().kind != TokenKind::Arrow) - { - error(peek().loc, std::format("expected '->' but found '{}'", peek().text)); - // Skip to end of line to recover. - while (peek().kind != TokenKind::Newline && peek().kind != TokenKind::EndOfInput) - { - next(); - } - break; - } - next(); // consume arrow - expectArrow = false; - continue; - } - - if (auto term = parseTerm()) - { - terms.push_back(std::move(*term)); - expectArrow = true; - } - else - { - break; - } - } - return terms; - } - - std::optional parseTerm() - { - const Token& token = peek(); - - if (token.kind == TokenKind::LParen) - { - return parseGroup(); - } - if (token.kind == TokenKind::Identifier) - { - return parseNamedTerm(); - } - - error(token.loc, std::format("expected an edge, stage, or group but found '{}'", token.text)); - next(); - return std::nullopt; - } - - Term parseGroup() - { - Term term; - term.kind = Term::Kind::Group; - term.loc = peek().loc; - next(); // ( - - while (peek().kind != TokenKind::RParen && peek().kind != TokenKind::Newline - && peek().kind != TokenKind::EndOfInput) - { - if (peek().kind == TokenKind::Identifier) - { - term.edges.push_back(peek().text); - next(); - } - else if (peek().kind == TokenKind::Comma) - { - next(); - } - else - { - error(peek().loc, std::format("expected an edge name in group but found '{}'", peek().text)); - next(); - } - } - if (peek().kind == TokenKind::RParen) - { - next(); - } - else - { - error(term.loc, "unterminated fan-in group; missing ')'"); - } - return term; - } - - Term parseNamedTerm() - { - Term term; - term.loc = peek().loc; - term.name = peek().text; - next(); - - // Optional `.key` (currently only meaningful for `out`). - if (peek().kind == TokenKind::Dot) - { - next(); - if (peek().kind == TokenKind::Identifier) - { - term.key = peek().text; - next(); - } - else - { - error(peek().loc, "expected a key name after '.'"); - } - } - - // A trailing `(...)` makes this a stage application with config args. - if (peek().kind == TokenKind::LParen) - { - term.kind = Term::Kind::Stage; - term.config = parseConfigArgs(); - return term; - } - - if (isReserved(term.name)) - { - term.kind = (term.name == "in") ? Term::Kind::Edge : Term::Kind::Boundary; - } - else - { - // Positional classification (edge vs stage) is resolved later; mark - // as Edge here and let buildStatement reinterpret odd positions. - term.kind = Term::Kind::Edge; - } - return term; - } - - nlohmann::json parseConfigArgs() - { - nlohmann::json config = nlohmann::json::object(); - next(); // ( - - while (peek().kind != TokenKind::RParen && peek().kind != TokenKind::Newline - && peek().kind != TokenKind::EndOfInput) - { - if (peek().kind == TokenKind::Comma) - { - next(); - continue; - } - if (peek().kind != TokenKind::Identifier) - { - error(peek().loc, std::format("expected a config key but found '{}'", peek().text)); - next(); - continue; - } - const std::string key = peek().text; - next(); - if (peek().kind != TokenKind::Equals) - { - error(peek().loc, std::format("expected '=' after config key '{}'", key)); - continue; - } - next(); // = - config[key] = parseConfigValue(); - } - if (peek().kind == TokenKind::RParen) - { - next(); - } - else - { - error(peek().loc, "unterminated config argument list; missing ')'"); - } - return config; - } - - nlohmann::json parseConfigValue() - { - const Token& token = peek(); - if (token.kind == TokenKind::String) - { - next(); - return token.text; - } - if (token.kind == TokenKind::Number) - { - next(); - if (token.text.find('.') != std::string::npos) - { - return std::stod(token.text); - } - return static_cast(std::stoll(token.text)); - } - if (token.kind == TokenKind::Identifier) - { - next(); - if (token.text == "true") - { - return true; - } - if (token.text == "false") - { - return false; - } - return token.text; // bareword treated as string - } - error(token.loc, std::format("expected a config value but found '{}'", token.text)); - next(); - return nullptr; - } - - std::vector mTokens; - std::size_t mPos = 0; - std::vector mDiagnostics; -}; - // ------------------------------ validation ------------------------------- inline void validateProgram(GraphProgram& program) @@ -836,46 +553,469 @@ inline void validateProgram(GraphProgram& program) } } +// ------------------------------- parser ---------------------------------- +// +// The DSL is newline-significant, so parseGraphProgram runs a lexy scanner over +// each line. Syntax errors are raised through lexy's error callback, which +// records them as located TextDiagnostics; a syntax error ends its statement, +// so each line reports at most one. + +namespace lexy_impl { + +namespace ld = lexy::dsl; + +// Identifier: [A-Za-z_][A-Za-z0-9_]* +inline constexpr auto identToken = + ld::token(ld::ascii::alpha_underscore + ld::while_(ld::ascii::alpha_digit_underscore)); + +// A run of spaces/tabs (and a stray CR) between tokens. +inline constexpr auto blankToken = ld::token(ld::while_(ld::ascii::blank / ld::lit_c<'\r'>)); + +// Ends a chunk of string contents: the closing quote or the start of an escape. +inline constexpr auto stringStop = ld::literal_set(ld::lit_c<'"'>, ld::lit_c<'\\'>); + +// The tag of every syntax error raised by this parser. Its message is prepared +// right before the error is raised (LineErrors::pendingMessage). +struct syntax_error +{ + static constexpr auto name = "syntax error"; +}; + +// Records the errors lexy reports for one line as located diagnostics. +struct LineErrors +{ + const char* lineBegin; + std::size_t lineNo; + std::vector* diagnostics; + std::string pendingMessage; + + void record(const char* position, std::string message) + { + const auto column = static_cast(position - lineBegin) + 1; + diagnostics->push_back({SourceLoc{lineNo, column}, std::move(message)}); + } +}; + +// The lexy error callback. Syntax errors carry their prepared message; lexy's +// own error kinds are formatted generically, so no error is ever reported +// without a location. +struct ErrorCallback +{ + using return_type = void; + + LineErrors* errors; + + template + void operator()(const lexy::error_context&, const lexy::error& error) const + { + std::string message = errors->pendingMessage.empty() ? std::string(error.message()) + : std::exchange(errors->pendingMessage, std::string{}); + errors->record(error.position(), std::move(message)); + } + + template + void operator()(const lexy::error_context&, const lexy::error& error) const + { + errors->record(error.position(), + std::format("expected '{}'", std::string_view(error.string(), error.length()))); + } + + template + void operator()(const lexy::error_context&, const lexy::error& error) const + { + errors->record(error.position(), + std::format("expected keyword '{}'", std::string_view(error.string(), error.length()))); + } + + template + void operator()(const lexy::error_context&, + const lexy::error& error) const + { + errors->record(error.position(), std::format("expected {}", error.name())); + } +}; + +// Parses one line into a flat term list. Returns false after reporting a syntax +// error, in which case the statement must not be built. +inline bool parseStatementLine(std::string_view lineText, + std::size_t lineNo, + std::vector& terms, + std::vector& diagnostics) +{ + LineErrors errors{lineText.data(), lineNo, &diagnostics, {}}; + auto input = lexy::string_input(lineText.data(), lineText.size()); + auto sc = lexy::scan(input, ErrorCallback{&errors}); + + const char* const lineEnd = lineText.data() + lineText.size(); + + auto locOf = [&](const char* position) { + return SourceLoc{lineNo, static_cast(position - lineText.data()) + 1}; + }; + auto rest = [&] { + return std::string_view(sc.position(), static_cast(lineEnd - sc.position())); + }; + auto skipBlank = [&] { + sc.parse(blankToken); + }; + auto atEndOfLine = [&] { + return sc.is_at_eof() || sc.peek(ld::lit_c<'#'>); + }; + auto captureIdent = [&] { + auto lexeme = sc.capture(identToken).value(); + return std::string(lexeme.begin(), lexeme.end()); + }; + + // Raises a syntax error at `position`; scanning stops. + auto fail = [&](const char* position, std::string message) { + errors.pendingMessage = std::move(message); + sc.fatal_error(syntax_error{}, position); + return false; + }; + + // Reports what was found at the current position instead of `expected`, + // or why the input there is invalid. + auto unexpected = [&](std::string_view expected) { + const Found found = describeFound(rest()); + return fail(sc.position(), found.invalid ? found.text : std::format("{} but found {}", expected, found.text)); + }; + + auto parseValue = [&](nlohmann::json& value) { + const char* const start = sc.position(); + const std::string_view text = rest(); + + if (sc.peek(ld::lit_c<'"'>)) + { + auto literal = scanStringLiteral(text); + if (!literal) + { + return fail(start, "unterminated string literal"); + } + // Consume chunk by chunk up to the closing quote; an escaped quote + // or backslash is consumed on its own so it cannot end a chunk. + sc.parse(ld::lit_c<'"'>); + while (sc) + { + sc.parse(ld::until(stringStop)); + if (!sc || sc.position()[-1] == '"') + { + break; + } + if (!sc.branch(ld::lit_c<'"'>)) + { + sc.branch(ld::lit_c<'\\'>); + } + } + value = std::move(literal->value); + return static_cast(sc); + } + + if (const std::size_t length = numberLength(text)) + { + std::string error; + auto number = numberValue(text.substr(0, length), error); + if (!number) + { + return fail(start, std::move(error)); + } + for (std::size_t i = 0; i < length; ++i) + { + sc.parse(ld::ascii::character); + } + value = std::move(*number); + return true; + } + + if (sc.peek(ld::ascii::alpha_underscore)) + { + const std::string word = captureIdent(); + value = word == "true" ? nlohmann::json(true) : word == "false" ? nlohmann::json(false) : nlohmann::json(word); + return true; + } + + return unexpected("expected an argument value"); + }; + + auto parseArgs = [&](nlohmann::json& config) { + const char* const open = sc.position(); + sc.parse(ld::lit_c<'('>); + config = nlohmann::json::object(); + while (true) + { + skipBlank(); + if (sc.branch(ld::lit_c<')'>)) + { + return true; + } + if (atEndOfLine()) + { + return fail(open, "unterminated argument list; missing ')'"); + } + if (sc.branch(ld::lit_c<','>)) + { + continue; + } + if (!sc.peek(ld::ascii::alpha_underscore)) + { + return unexpected("expected an argument name"); + } + const std::string key = captureIdent(); + skipBlank(); + if (!sc.branch(ld::lit_c<'='>)) + { + return unexpected(std::format("expected '=' after argument '{}'", key)); + } + skipBlank(); + nlohmann::json value; + if (!parseValue(value)) + { + return false; + } + config[key] = std::move(value); + } + }; + + auto parseGroup = [&] { + const char* const open = sc.position(); + Term term; + term.kind = Term::Kind::Group; + term.loc = locOf(open); + sc.parse(ld::lit_c<'('>); + while (true) + { + skipBlank(); + if (sc.branch(ld::lit_c<')'>)) + { + terms.push_back(std::move(term)); + return true; + } + if (atEndOfLine()) + { + return fail(open, "unterminated fan-in group; missing ')'"); + } + if (sc.branch(ld::lit_c<','>)) + { + continue; + } + if (!sc.peek(ld::ascii::alpha_underscore)) + { + return unexpected("expected an edge name in group"); + } + term.edges.push_back(captureIdent()); + } + }; + + auto parseNamed = [&] { + Term term; + term.loc = locOf(sc.position()); + term.name = captureIdent(); + + // Optional `.key` (only meaningful for `out`). + skipBlank(); + if (sc.branch(ld::lit_c<'.'>)) + { + skipBlank(); + if (!sc.peek(ld::ascii::alpha_underscore)) + { + return unexpected("expected a key name after '.'"); + } + term.key = captureIdent(); + skipBlank(); + } + + // A trailing `(...)` makes this a stage application with arguments. + if (sc.peek(ld::lit_c<'('>)) + { + term.kind = Term::Kind::Stage; + if (!parseArgs(term.config)) + { + return false; + } + } + else if (isReserved(term.name)) + { + term.kind = (term.name == "in") ? Term::Kind::Edge : Term::Kind::Boundary; + } + else + { + // Positional edge/stage classification is resolved by buildStatement. + term.kind = Term::Kind::Edge; + } + terms.push_back(std::move(term)); + return true; + }; + + auto parseTerm = [&] { + skipBlank(); + if (sc.peek(ld::lit_c<'('>)) + { + return parseGroup(); + } + if (sc.peek(ld::ascii::alpha_underscore)) + { + return parseNamed(); + } + return unexpected("expected an edge, stage, or group"); + }; + + skipBlank(); + if (atEndOfLine()) + { + return true; // blank or comment-only line + } + if (!parseTerm()) + { + return false; + } + while (true) + { + skipBlank(); + if (atEndOfLine()) + { + return true; + } + if (!sc.branch(LEXY_LIT("->"))) + { + return unexpected("expected '->'"); + } + if (!parseTerm()) + { + return false; + } + } +} + +} // namespace lexy_impl + } // namespace detail -// Parses and structurally validates a Model-B graph program from text. All -// problems (lexer, parser, and semantic) are collected into -// GraphProgram::diagnostics, each located by line/column. +// Parses and structurally validates a graph program from text. All problems +// (syntax and structure) are collected into GraphProgram::diagnostics, each +// located by line/column and sorted by location. inline GraphProgram parseGraphProgram(std::string_view text) { - detail::Lexer lexer(text); - auto tokens = lexer.tokenize(); + GraphProgram program; + std::size_t outputIndex = 0; + std::size_t lineNo = 0; + std::size_t pos = 0; - detail::Parser parser(std::move(tokens), lexer.errors()); - GraphProgram program = parser.parse(); + while (pos <= text.size()) + { + const std::size_t newline = text.find('\n', pos); + const std::string_view line = + (newline == std::string_view::npos) ? text.substr(pos) : text.substr(pos, newline - pos); + ++lineNo; + + std::vector terms; + if (detail::lexy_impl::parseStatementLine(line, lineNo, terms, program.diagnostics) && !terms.empty()) + { + detail::buildStatement(terms, program, outputIndex, program.diagnostics); + } + + if (newline == std::string_view::npos) + { + break; + } + pos = newline + 1; + } detail::validateProgram(program); + detail::sortDiagnostics(program.diagnostics); return program; } -// Renders a parsed program as a Mermaid flowchart so the DAG can be visualized. +namespace detail { + +// Escapes the characters that would end or confuse a quoted Mermaid label. +inline std::string mermaidLabel(std::string_view text) +{ + std::string label; + for (char c : text) + { + if (c == '"') + { + label += "#quot;"; + } + else if (c == '<') + { + label += "#lt;"; + } + else if (c == '>') + { + label += "#gt;"; + } + else + { + label += c; + } + } + return label; +} + +} // namespace detail + +// Renders a parsed program as a Mermaid flowchart: stages are boxes (listing +// their config), edges are rounded nodes, and every `-> end` gets its own +// terminal node. Node ids are generated, so edge names never clash with Mermaid +// keywords such as `end`. inline std::string toMermaid(const GraphProgram& program) { - std::string out = "flowchart LR\n"; + std::unordered_map outputLabels; + for (const auto& output : program.outputs) + { + outputLabels[output.edge] = output.key ? "out." + *output.key + : program.outputs.size() > 1 ? std::format("out[{}]", output.index) + : std::string{"out"}; + } - std::size_t nodeId = 0; - for (const auto& stage : program.stages) + std::string nodes; + std::string links; + std::unordered_map edgeIds; + + auto edgeId = [&](const std::string& edge) -> std::string { + if (auto it = edgeIds.find(edge); it != edgeIds.end()) + { + return it->second; + } + std::string id = std::format("e{}", edgeIds.size()); + auto label = outputLabels.find(edge); + nodes += std::format(" {}([\"{}\"])\n", id, + detail::mermaidLabel(label != outputLabels.end() ? label->second : edge)); + edgeIds.emplace(edge, id); + return id; + }; + + std::size_t deadEnds = 0; + for (std::size_t i = 0; i < program.stages.size(); ++i) { - const std::string id = std::format("n{}", nodeId++); + const StageNode& stage = program.stages[i]; + + std::string label = detail::mermaidLabel(stage.type); + if (stage.config.is_object()) + { + for (const auto& item : stage.config.items()) + { + const std::string value = item.value().is_string() ? item.value().get() : item.value().dump(); + label += std::format("
{}={}", detail::mermaidLabel(item.key()), detail::mermaidLabel(value)); + } + } + nodes += std::format(" s{}[\"{}\"]\n", i, label); + for (const auto& input : stage.inputs) { - out += std::format(" {}([{}]) --> {}[{}]\n", input, input, id, stage.type); + links += std::format(" {} --> s{}\n", edgeId(input), i); } if (stage.output) { - out += std::format(" {}[{}] --> {}([{}])\n", id, stage.type, *stage.output, *stage.output); + links += std::format(" s{} --> {}\n", i, edgeId(*stage.output)); } else { - out += std::format(" {}[{}] --> end([end])\n", id, stage.type); + nodes += std::format(" d{}[[\"end\"]]\n", deadEnds); + links += std::format(" s{} --> d{}\n", i, deadEnds); + ++deadEnds; } } - return out; + return "flowchart LR\n" + nodes + links; } } // namespace filterGraph::dsl diff --git a/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp b/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp new file mode 100644 index 0000000..b10d751 --- /dev/null +++ b/libs/filterGraph/core/filterGraph/GraphLangHandwritten.hpp @@ -0,0 +1,474 @@ +#pragma once + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +// DEPRECATED: the original hand-written tokenizer and recursive-descent parser +// for the graph DSL. dsl::parseGraphProgram (GraphLang.hpp) now uses the +// lexy-based parser, which produces the same IR and diagnostics; the tests +// check the two against each other. This parser will be removed in a future +// release. +namespace filterGraph::dsl { + +namespace detail::handwritten { + +enum class TokenKind +{ + Identifier, + Number, + String, + Arrow, // -> + LParen, // ( + RParen, // ) + Comma, // , + Equals, // = + Dot, // . + Invalid, // an unexpected character or an unterminated string + Newline, + EndOfInput +}; + +struct Token +{ + TokenKind kind; + std::string text; // source text; the unescaped value for strings + SourceLoc loc; + std::size_t offset = 0; // position in the source +}; + +class Lexer +{ +public: + explicit Lexer(std::string_view source) + : mSource(source) + { + } + + std::vector tokenize() + { + std::vector tokens; + while (mPos < mSource.size()) + { + const char c = mSource[mPos]; + const std::string_view rest = mSource.substr(mPos); + + if (c == '\n') + { + tokens.push_back(endOfLine(TokenKind::Newline)); + advance(1); + continue; + } + if (c == '\r' || c == ' ' || c == '\t') + { + advance(1); + continue; + } + if (c == '#') + { + mCommentStart = Mark{currentLoc(), mPos}; + advance(std::min(rest.find('\n'), rest.size())); + continue; + } + if (rest.starts_with("->")) + { + tokens.push_back(take(TokenKind::Arrow, 2)); + continue; + } + if (c == '(' || c == ')' || c == ',' || c == '=' || c == '.') + { + const TokenKind kind = c == '(' ? TokenKind::LParen + : c == ')' ? TokenKind::RParen + : c == ',' ? TokenKind::Comma + : c == '=' ? TokenKind::Equals + : TokenKind::Dot; + tokens.push_back(take(kind, 1)); + continue; + } + if (c == '"') + { + if (auto literal = scanStringLiteral(rest)) + { + Token token = take(TokenKind::String, literal->length); + token.text = std::move(literal->value); + tokens.push_back(std::move(token)); + } + else + { + tokens.push_back(take(TokenKind::Invalid, std::min(rest.find('\n'), rest.size()))); + } + continue; + } + if (const std::size_t length = numberLength(rest)) + { + tokens.push_back(take(TokenKind::Number, length)); + continue; + } + if (isIdentStart(c)) + { + std::size_t length = 1; + while (length < rest.size() && isIdentPart(rest[length])) + { + ++length; + } + tokens.push_back(take(TokenKind::Identifier, length)); + continue; + } + tokens.push_back(take(TokenKind::Invalid, 1)); + } + tokens.push_back(endOfLine(TokenKind::EndOfInput)); + return tokens; + } + +private: + struct Mark + { + SourceLoc loc; + std::size_t offset; + }; + + // A token of `length` characters starting at the current position. + Token take(TokenKind kind, std::size_t length) + { + Token token{kind, std::string(mSource.substr(mPos, length)), currentLoc(), mPos}; + advance(length); + return token; + } + + // Newline / end of input, located where the line's content ends: at its + // comment, if it has one. + Token endOfLine(TokenKind kind) + { + const Mark mark = mCommentStart.value_or(Mark{currentLoc(), mPos}); + mCommentStart.reset(); + return Token{kind, {}, mark.loc, mark.offset}; + } + + SourceLoc currentLoc() const + { + return SourceLoc{mLine, mColumn}; + } + + void advance(std::size_t count) + { + for (; count > 0 && mPos < mSource.size(); --count) + { + if (mSource[mPos] == '\n') + { + ++mLine; + mColumn = 1; + } + else + { + ++mColumn; + } + ++mPos; + } + } + + std::string_view mSource; + std::size_t mPos = 0; + std::size_t mLine = 1; + std::size_t mColumn = 1; + std::optional mCommentStart; +}; + +// Recursive-descent parser over the token stream. A syntax error ends its +// statement: the rest of the line is skipped and the statement is not built, +// so each line reports at most one syntax error. +class Parser +{ +public: + Parser(std::string_view source, std::vector tokens) + : mSource(source) + , mTokens(std::move(tokens)) + { + } + + GraphProgram parse() + { + GraphProgram program; + std::size_t outputIndex = 0; + + while (peek().kind != TokenKind::EndOfInput) + { + if (peek().kind == TokenKind::Newline) + { + next(); + continue; + } + + mFailed = false; + auto terms = parseStatement(); + if (mFailed) + { + while (!atEndOfLine()) + { + next(); + } + continue; + } + buildStatement(terms, program, outputIndex, mDiagnostics); + } + + program.diagnostics = std::move(mDiagnostics); + return program; + } + +private: + const Token& peek() const + { + return mTokens[mPos]; + } + + const Token& next() + { + return mTokens[mPos < mTokens.size() - 1 ? mPos++ : mPos]; + } + + bool atEndOfLine() const + { + return peek().kind == TokenKind::Newline || peek().kind == TokenKind::EndOfInput; + } + + void error(const SourceLoc& loc, std::string message) + { + mDiagnostics.push_back({loc, std::move(message)}); + mFailed = true; + } + + // Reports what was found at the current token instead of `expected`, or + // why the input there is invalid. + void unexpected(std::string_view expected) + { + const Found found = describeFound(mSource.substr(peek().offset)); + error(peek().loc, found.invalid ? found.text : std::format("{} but found {}", expected, found.text)); + } + + // Parses a single line into a flat list of terms. + std::vector parseStatement() + { + std::vector terms; + while (true) + { + Term term = parseTerm(); + if (mFailed) + { + return terms; + } + terms.push_back(std::move(term)); + if (atEndOfLine()) + { + return terms; + } + if (peek().kind != TokenKind::Arrow) + { + unexpected("expected '->'"); + return terms; + } + next(); + } + } + + Term parseTerm() + { + if (peek().kind == TokenKind::LParen) + { + return parseGroup(); + } + if (peek().kind == TokenKind::Identifier) + { + return parseNamedTerm(); + } + unexpected("expected an edge, stage, or group"); + return {}; + } + + Term parseGroup() + { + Term term; + term.kind = Term::Kind::Group; + term.loc = peek().loc; + next(); // ( + + while (true) + { + if (peek().kind == TokenKind::RParen) + { + next(); + return term; + } + if (atEndOfLine()) + { + error(term.loc, "unterminated fan-in group; missing ')'"); + return term; + } + if (peek().kind == TokenKind::Comma) + { + next(); + continue; + } + if (peek().kind != TokenKind::Identifier) + { + unexpected("expected an edge name in group"); + return term; + } + term.edges.push_back(peek().text); + next(); + } + } + + Term parseNamedTerm() + { + Term term; + term.loc = peek().loc; + term.name = peek().text; + next(); + + // Optional `.key` (currently only meaningful for `out`). + if (peek().kind == TokenKind::Dot) + { + next(); + if (peek().kind != TokenKind::Identifier) + { + unexpected("expected a key name after '.'"); + return term; + } + term.key = peek().text; + next(); + } + + // A trailing `(...)` makes this a stage application with arguments. + if (peek().kind == TokenKind::LParen) + { + term.kind = Term::Kind::Stage; + term.config = parseArguments(); + return term; + } + + if (isReserved(term.name)) + { + term.kind = (term.name == "in") ? Term::Kind::Edge : Term::Kind::Boundary; + } + else + { + // Positional classification (edge vs stage) is resolved later; mark + // as Edge here and let buildStatement reinterpret odd positions. + term.kind = Term::Kind::Edge; + } + return term; + } + + nlohmann::json parseArguments() + { + nlohmann::json config = nlohmann::json::object(); + const SourceLoc open = peek().loc; + next(); // ( + + while (true) + { + if (peek().kind == TokenKind::RParen) + { + next(); + return config; + } + if (atEndOfLine()) + { + error(open, "unterminated argument list; missing ')'"); + return config; + } + if (peek().kind == TokenKind::Comma) + { + next(); + continue; + } + if (peek().kind != TokenKind::Identifier) + { + unexpected("expected an argument name"); + return config; + } + const std::string key = peek().text; + next(); + if (peek().kind != TokenKind::Equals) + { + unexpected(std::format("expected '=' after argument '{}'", key)); + return config; + } + next(); // = + nlohmann::json value = parseValue(); + if (mFailed) + { + return config; + } + config[key] = std::move(value); + } + } + + nlohmann::json parseValue() + { + const Token& token = peek(); + if (token.kind == TokenKind::String) + { + next(); + return token.text; + } + if (token.kind == TokenKind::Number) + { + std::string numberError; + auto number = numberValue(token.text, numberError); + if (!number) + { + error(token.loc, std::move(numberError)); + return nullptr; + } + next(); + return std::move(*number); + } + if (token.kind == TokenKind::Identifier) + { + next(); + if (token.text == "true") + { + return true; + } + if (token.text == "false") + { + return false; + } + return token.text; // bareword treated as string + } + unexpected("expected an argument value"); + return nullptr; + } + + std::string_view mSource; + std::vector mTokens; + std::size_t mPos = 0; + bool mFailed = false; // the current statement has a syntax error + std::vector mDiagnostics; +}; + +} // namespace detail::handwritten + +// Parses a graph program with the deprecated hand-written parser. Produces the +// same result as parseGraphProgram. +[[deprecated("the hand-written DSL parser is deprecated; use dsl::parseGraphProgram, which gives the same result")]] +inline GraphProgram parseGraphProgramHandwritten(std::string_view text) +{ + detail::handwritten::Parser parser(text, detail::handwritten::Lexer(text).tokenize()); + GraphProgram program = parser.parse(); + + detail::validateProgram(program); + detail::sortDiagnostics(program.diagnostics); + return program; +} + +} // namespace filterGraph::dsl diff --git a/libs/filterGraph/core/filterGraph/GraphLangLexy.hpp b/libs/filterGraph/core/filterGraph/GraphLangLexy.hpp index 72c5bd3..c0c9ae5 100644 --- a/libs/filterGraph/core/filterGraph/GraphLangLexy.hpp +++ b/libs/filterGraph/core/filterGraph/GraphLangLexy.hpp @@ -2,305 +2,17 @@ #include -#include -#include -#include -#include - -#include - -#include -#include -#include -#include #include -#include -// Prototype: a lexy-based parser producing the SAME GraphProgram IR as the -// hand-written parser in GraphLang.hpp, kept alongside it for comparison. -// -// The DSL is newline-significant, so the C++ driver splits the input into -// lines (stripping `#` comments and blank lines) and runs a lexy scanner over -// each statement. The scanner uses lexy's imperative scanning interface, which -// classifies each term exactly like `parseNamedTerm`/`parseGroup` in -// GraphLang.hpp so the shared `buildStatement`/`validateProgram` helpers yield -// identical output. +// The lexy-based parser now lives in GraphLang.hpp and backs +// dsl::parseGraphProgram. This header keeps the name introduced in 0.2.0 +// working. namespace filterGraph::dsl { -namespace detail { -namespace lexy_impl { - -namespace ld = lexy::dsl; - -// Identifier: [A-Za-z_][A-Za-z0-9_]* as a capturable token. -inline constexpr auto identToken = - ld::token(ld::ascii::alpha_underscore + ld::while_(ld::ascii::alpha_digit_underscore)); - -// A run of spaces/tabs (and a stray CR) to discard between tokens. -inline constexpr auto blankToken = ld::token(ld::while_(ld::ascii::blank / ld::lit_c<'\r'>)); - -// Parses one DSL statement line into a flat term list. Returns false on a -// syntax error (the caller then records a generic "syntax error" diagnostic). -inline bool parseStatementLine(std::string_view lineText, std::size_t lineNo, std::vector& terms) -{ - auto input = lexy::string_input(lineText.data(), lineText.size()); - auto sc = lexy::scan(input, lexy::noop); - - const SourceLoc loc{lineNo, 1}; - - auto skipBlank = [&] { sc.discard(blankToken); }; - - auto captureIdent = [&]() -> std::optional { - auto result = sc.capture(identToken); - if (!result) - { - return std::nullopt; - } - auto lexeme = result.value(); - return std::string(lexeme.begin(), lexeme.end()); - }; - - auto parseValue = [&]() -> nlohmann::json { - skipBlank(); - if (sc.peek(ld::lit_c<'"'>)) - { - sc.parse(ld::lit_c<'"'>); - auto result = sc.capture(ld::token(ld::while_(ld::ascii::character - ld::lit_c<'"'>))); - sc.parse(ld::lit_c<'"'>); - if (!result) - { - return nullptr; - } - auto lexeme = result.value(); - return std::string(lexeme.begin(), lexeme.end()); - } - if (sc.peek(ld::lit_c<'-'>) || sc.peek(ld::ascii::digit)) - { - auto result = sc.capture(ld::token(ld::if_(ld::lit_c<'-'>) + ld::digits<>)); - if (!result) - { - return nullptr; - } - auto lexeme = result.value(); - const std::string text(lexeme.begin(), lexeme.end()); - return static_cast(std::stoll(text)); - } - // Bareword: true/false become bools, anything else a string. - auto word = captureIdent(); - if (!word) - { - return nullptr; - } - if (*word == "true") - { - return true; - } - if (*word == "false") - { - return false; - } - return *word; - }; - - auto parseConfigArgs = [&]() -> nlohmann::json { - nlohmann::json config = nlohmann::json::object(); - sc.parse(ld::lit_c<'('>); - while (true) - { - skipBlank(); - if (sc.peek(ld::lit_c<')'>) || sc.is_at_eof()) - { - break; - } - auto key = captureIdent(); - if (!key) - { - sc.fatal_error("syntax error", sc.position()); - break; - } - skipBlank(); - if (!sc.branch(ld::lit_c<'='>)) - { - sc.fatal_error("syntax error", sc.position()); - break; - } - config[*key] = parseValue(); - skipBlank(); - sc.branch(ld::lit_c<','>); // optional separator - } - sc.parse(ld::lit_c<')'>); - return config; - }; - - auto parseGroup = [&]() -> bool { - Term term; - term.kind = Term::Kind::Group; - term.loc = loc; - sc.parse(ld::lit_c<'('>); - while (true) - { - skipBlank(); - if (sc.peek(ld::lit_c<')'>) || sc.is_at_eof()) - { - break; - } - if (sc.branch(ld::lit_c<','>)) - { - continue; - } - auto edge = captureIdent(); - if (!edge) - { - sc.fatal_error("syntax error", sc.position()); - return false; - } - term.edges.push_back(std::move(*edge)); - } - if (!sc.branch(ld::lit_c<')'>)) - { - sc.fatal_error("syntax error", sc.position()); - return false; - } - terms.push_back(std::move(term)); - return true; - }; - - auto parseNamed = [&]() -> bool { - Term term; - term.loc = loc; - auto name = captureIdent(); - if (!name) - { - sc.fatal_error("syntax error", sc.position()); - return false; - } - term.name = std::move(*name); - - // Optional `.key` (only meaningful for `out`). - if (sc.peek(ld::lit_c<'.'>)) - { - sc.parse(ld::lit_c<'.'>); - auto key = captureIdent(); - if (!key) - { - sc.fatal_error("syntax error", sc.position()); - return false; - } - term.key = std::move(*key); - } - - // A trailing `(...)` makes this a stage application with config args. - skipBlank(); - if (sc.peek(ld::lit_c<'('>)) - { - term.kind = Term::Kind::Stage; - term.config = parseConfigArgs(); - terms.push_back(std::move(term)); - return true; - } - - if (isReserved(term.name)) - { - term.kind = (term.name == "in") ? Term::Kind::Edge : Term::Kind::Boundary; - } - else - { - // Positional edge/stage classification is resolved by buildStatement. - term.kind = Term::Kind::Edge; - } - terms.push_back(std::move(term)); - return true; - }; - - auto parseTerm = [&]() -> bool { - skipBlank(); - if (sc.peek(ld::lit_c<'('>)) - { - return parseGroup(); - } - if (sc.peek(ld::ascii::alpha_underscore)) - { - return parseNamed(); - } - sc.fatal_error("syntax error", sc.position()); - return false; - }; - - skipBlank(); - if (sc.is_at_eof()) - { - return true; // empty statement - } - if (!parseTerm()) - { - return false; - } - while (true) - { - skipBlank(); - if (sc.is_at_eof()) - { - break; - } - if (!sc.branch(LEXY_LIT("->"))) - { - sc.fatal_error("syntax error", sc.position()); - return false; - } - if (!parseTerm()) - { - return false; - } - } - return static_cast(sc); -} - -} // namespace lexy_impl -} // namespace detail - -// Parses and structurally validates a Model-B graph program from text using a -// lexy grammar, producing the SAME IR as parseGraphProgram. +// Same as parseGraphProgram. inline GraphProgram parseGraphProgramLexy(std::string_view text) { - GraphProgram program; - std::size_t outputIndex = 0; - std::size_t lineNo = 0; - std::size_t pos = 0; - - while (pos <= text.size()) - { - const std::size_t newline = text.find('\n', pos); - const std::string_view raw = - (newline == std::string_view::npos) ? text.substr(pos) : text.substr(pos, newline - pos); - ++lineNo; - - // Strip a `#` comment to end-of-line. - const std::size_t hash = raw.find('#'); - std::string_view line = (hash == std::string_view::npos) ? raw : raw.substr(0, hash); - - const bool blank = line.find_first_not_of(" \t\r") == std::string_view::npos; - if (!blank) - { - std::vector terms; - if (detail::lexy_impl::parseStatementLine(line, lineNo, terms) && !terms.empty()) - { - detail::buildStatement(terms, program, outputIndex, program.diagnostics); - } - else - { - program.diagnostics.push_back({SourceLoc{lineNo, 1}, "syntax error"}); - } - } - - if (newline == std::string_view::npos) - { - break; - } - pos = newline + 1; - } - - detail::validateProgram(program); - return program; + return parseGraphProgram(text); } } // namespace filterGraph::dsl diff --git a/libs/filterGraph/core/filterGraph/MergeFilter.hpp b/libs/filterGraph/core/filterGraph/MergeFilter.hpp new file mode 100644 index 0000000..6789ec8 --- /dev/null +++ b/libs/filterGraph/core/filterGraph/MergeFilter.hpp @@ -0,0 +1,65 @@ +#pragma once + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace filterGraph { + +// The input of a merge stage: one slot per fan-in edge, in the order the edges +// are listed in the DSL group `(a, b, c) -> Merge`. An empty slot is a "hole" +// left by an upstream path that dropped the message via std::nullopt. +using MergeInputs = std::vector; + +// MergeFilter is the fan-in stage of the text DSL: it combines the +// values arriving on several edges into a single OutputType via a +// caller-supplied combiner. It is an ordinary MessageFilter, so it lives in the same FilterRegistry as every other stage; the +// DSL runtime recognises a merge by its MergeInputs input type and gathers the +// 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. +template +class MergeFilter : public MessageFilter +{ +public: + using Combiner = std::function(MergeInputs&&)>; + + explicit MergeFilter(Combiner combiner) + : mCombiner(std::move(combiner)) + { + } + + std::optional filter(MergeInputs&& inputs) override + { + return mCombiner(std::move(inputs)); + } + +private: + Combiner mCombiner; +}; + +// Registers MergeFilter under `name` for use as a DSL fan-in stage: +// +// (a, b) -> Name -> merged +template +void registerMergeFilter(const std::string& name, typename MergeFilter::Combiner combiner) +{ + FilterRegistrar> registrar( + name, + [combiner = std::move(combiner)](const nlohmann::json&) { + return std::make_shared>(combiner); + }); +} + +} // namespace filterGraph diff --git a/tests/FilterGraphTests/FilterGraphTests.cpp b/tests/FilterGraphTests/FilterGraphTests.cpp index 71e547f..7c26489 100644 --- a/tests/FilterGraphTests/FilterGraphTests.cpp +++ b/tests/FilterGraphTests/FilterGraphTests.cpp @@ -1,20 +1,27 @@ #include +#include #include #include #include +#include #include #include #include #include +#include #include #include #include #include +#include +#include #include #include #include +#include +#include using namespace filterGraph; @@ -408,8 +415,7 @@ TEST_CASE("parseGraphProgram handles fan-in via a group", "[GraphDsl]") static FilterRegistrar registerDslP1("DslP1"); static FilterRegistrar registerDslP2("DslP2"); static const bool registerDslMerge = [] { - registerJoinFilter("DslMerge", - [](std::vector&&) -> std::optional { return 0; }); + registerMergeFilter("DslMerge", [](MergeInputs&&) -> std::optional { return 0; }); return true; }(); (void)registerDslMerge; @@ -561,8 +567,7 @@ TEST_CASE("parseGraphProgramLexy handles fan-in via a group", "[GraphDslLexy]") static FilterRegistrar registerLexyP1("LexyP1"); static FilterRegistrar registerLexyP2("LexyP2"); static const bool registerLexyMerge = [] { - registerJoinFilter("LexyMerge", - [](std::vector&&) -> std::optional { return 0; }); + registerMergeFilter("LexyMerge", [](MergeInputs&&) -> std::optional { return 0; }); return true; }(); (void)registerLexyMerge; @@ -632,3 +637,398 @@ TEST_CASE("parseGraphProgramLexy matches parseGraphProgram on the same input", " REQUIRE(lexy.stages.size() == handWritten.stages.size()); REQUIRE(lexy.outputs.size() == handWritten.outputs.size()); } + +// --------------------------------------------------------------------------- +// DslFilterGraph: running graphs described in the text DSL +// --------------------------------------------------------------------------- + +namespace { + +std::vector& runLog() +{ + static std::vector log; + return log; +} + +// Appends "