Crate wingfoil

Expand description

A Rust stream-processing library: build a directed acyclic graph of data transformations once, then run it against live data or replay it over history with identical semantics.

use std::time::Duration;
use wingfoil::prelude::*;
use wingfoil::{NanoTime, RunFor, RunMode};

let g = GraphBuilder::new();
let count = g.ticker(Duration::from_millis(10)).count();
let is_even = count.map(|n: &u64| n.is_multiple_of(2));
let total = count.filter(&is_even).fold(0u64, |acc, v| *acc += v);

let mut runner = g.build();
runner.run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Cycles(10)).unwrap();
assert_eq!(runner.value(&total), 2 + 4 + 6 + 8 + 10);

Swap RunMode::HistoricalFrom for RunMode::RealTime and the same graph runs against the wall clock — that is the point of the two run modes.

§The idea: node semantics as a function, not an object

An Op defines what a node does as a pure associated function — cycle(cfg, state, input, ctx) — over engine-owned state and typed inputs the engine passes in, with a const ACTIVATION declaring how it is scheduled. Nothing about a node’s computation is tied to how its storage is allocated or how its inputs are fetched.

That separation is what lets one definition drive several execution strategies, and it is the property the whole design is arranged around:

the interpreted engine (interp::Builder) owns the value slots and the state, adapts each Op behind a single dyn boundary, and drives the Kernel;
a compiled runner is a plain function with node state in local variables, calling the same Op::cycle functions monomorphized, with tick propagation as bools the optimiser can see through;
a nested island packs a whole sub-graph into one node of an interpreted graph, running that same compiled code inside it.

Every one of those executes the identical semantics code. There is no duplicated cycle logic anywhere — no per-kind emitter strings, no cycle_inline twins — so the strategies cannot drift from each other. nitro! is the front door: one wiring function in, all three out.

§The tour

nitro! — one wiring function, three execution paths from the same tokens: interpreted(), fully-monomorphized compiled(), and nested() (a compiled island as one node of an interpreted graph).
fluent — the chaining API as extension traits (SourceOps for sources, StreamOps for combinators), so the op vocabulary is open; prelude brings the common set into scope.
ops — the op catalog (map/filter/fold/join/delay/window/… plus the sources), and adapters::statistics — EWMA and rolling-window statistics as a separate opt-in StatisticsOps trait. An op is single-sourced through one mechanism: #[op(build = name)] on its Op impl generates the interpreted Builder method and the nitro! forwarder functions every compiled/nested emission dispatches through, both derived from the op’s declared shape — there is no per-op table in the macro, so built-in and user ops take the identical path.
Sources in every activation mode: Activation::THREADED external, busy-spin Activation::ALWAYS poll, the both-modes channel, and feedback edges. All non-coalescing: same-instant values ride one Burst, never latest-wins.
adapters — the I/O surface (CSV, Kafka, ZeroMQ, KDB+, Redis, Postgres, etcd, FIX, web, Aeron, iceoryx2, Fluvio, augurs, Prometheus, OTLP), each behind its own feature and kept out of the prelude — opt in with use wingfoil::adapters::<name>::….
latency — stamp wall-clock timestamps onto messages as they hop through ops and across processes, then aggregate per-stage deltas.
introspect — the wired topology as data and as pictures (Graphviz / Mermaid / JSON / GML), from GraphBuilder::snapshot or Runner::snapshot. Active and passive edges are drawn differently, which is usually the reason to want the picture.
channel — the Message envelope and senders; async_source (the async feature) wraps it as produce_async, an async producer of timestamped values that replays deterministically in historical mode.
pool — recycled payload buffers behind cheap non-atomic Pooled handles, and the loan-based pooled_channel producer API: zero payload allocations at steady state, with the bounded pool as backpressure.
Fallible lifecycle — every Op function returns anyhow::Result; the interpreted Runner reports the first start/cycle/stop/teardown error with node context and still runs cleanup.

For how the pieces fit together, and why, see docs/wingfoil-architecture.md.

§Threading: a graph lives on one thread

GraphBuilder, Stream<T> and Runner are !Send and !Sync. They hold Rc internally, so wiring, build and run all have to happen on the same thread, and moving any of them into std::thread::spawn is a compile error along the lines of:

error[E0277]: `Rc<RefCell<wingfoil::interp::Builder>>` cannot be sent
              between threads safely
  = help: within `{closure@...}`, the trait `Send` is not implemented for
          `Rc<RefCell<wingfoil::interp::Builder>>`
note: required because it appears within the type
      `wingfoil::fluent::GraphBuilder`

(Whichever of the three you moved: Stream<T> names the same Rc, and Runner names the Rc slots it owns.)

That is a deliberate contract, not a missing impl. Node state is RefCell-owned by the engine and read back through Rc slots precisely so that no lock is ever taken on the graph execution path — a mutex inside cycle would be a correctness problem as much as a cost one. Making the wiring types Send would only move the synchronisation somewhere less visible.

Nothing stops you running several graphs, one per thread. What you cannot do is share one.

§Crossing the thread boundary

Every supported crossing hands you a Send half while the graph stays where it is:

You want to…	Wire	The `Send` half to move
feed a graph from a thread, socket or async task	`channel`	`ChannelSender<T>`
…the same, with recycled buffers and no payload allocation	`pooled_channel`	`PooledSender<T>`
…the same, but connect at run start rather than at wiring	`source_at_start`	`ChannelSender<T>`, handed to your `setup`
run a whole producer sub-graph on a worker thread	`spawn`	nothing — the worker wires its own graph
offload one stage of an existing pipeline	`spawn_map`	nothing — ditto, in lock-step
push values in from realtime code, minimal envelope	`external`	`ExternalSource<T>`

channel is the general answer, and the one the I/O adapters are built on. The sender is Send and clonable, each send wakes the realtime kernel, and send_at stamps a value so a historical replay stays deterministic:

use std::thread;
use std::time::Duration;
use wingfoil::prelude::*;
use wingfoil::{NanoTime, RunFor, RunMode};

let g = GraphBuilder::new();
// The `Stream` stays on this thread; the `ChannelSender` is the `Send` half.
let (values, sender) = g.channel::<u64>();
let total = values
    .map(|b: &Burst<u64>| b.iter().sum::<u64>())
    .fold(0u64, |acc, v| *acc += v);
let mut runner = g.build();

let producer = thread::spawn(move || {
    for i in 1..=3u64 {
        sender.send_at(i, NanoTime::from(Duration::from_secs(i)));
    }
    // End-of-stream: the receiving graph winds down once it has drained.
    sender.close();
});

runner.run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever).unwrap();
producer.join().unwrap();
assert_eq!(runner.value(&total), 1 + 2 + 3);

Note what did not cross: g, values, total and runner all stayed on the calling thread. If the producer is itself a wingfoil graph, reach for spawn instead and it will wire and run one on the worker for you — see examples/core/threading and examples/core/spawn.

§Tracing and instrumentation

The engine can emit tracing spans around its own execution. Every span site is behind its own feature, and the tracing dependency itself is optional, so a default build carries neither the dependency nor a single span:

feature	what it adds
`tracing`	The `tracing` dependency. On its own it emits nothing — enable one of the below.
`instrument-run`	A span around `Runner::run` (and `run_dynamic`, under `dynamic-graph`) — the whole start→cycles→stop→teardown lifecycle.
`instrument-cycle`	A span around each engine cycle (one per dirty-node batch).
`instrument-apply-nodes`	A span around each lifecycle phase (start / stop / teardown) applied over all nodes, recording the phase in `desc`.
`instrument-initialise`	A span around graph initialisation (`Builder::build`).
`instrument-cycle-node`	A span per node execution, recording the node index and label. High frequency — opt in deliberately.
`instrument-default`	`instrument-run` + `instrument-cycle` + `instrument-apply-nodes` + `instrument-initialise`.
`instrument-all`	`instrument-default` plus `instrument-cycle-node`.

All instrument-* features imply tracing. Both dispatch strategies (the sparse drain and the FullSweep oracle) emit the same spans, so instrumentation cannot tell them apart — just as results cannot. See examples/core/tracing for a runnable demonstration, and StreamOps::logged for the per-value debug tap (which emits through the log crate, independently of these features).

§Known limits

Documented, not forgotten: merge/join are fixed at two inputs on the compiled path (the interpreted side has variadic merge_n); the interpreted value store is per-node slots rather than an arena/SoA, and its dirty list is drained rather than topologically ordered; and compiled() is a closed box — static topology, outputs only, no I/O or live inputs, by design. See docs/planning/port-plan.md “Deferred / post-v1 work”.

Re-exports§

pub use log;

Modules§

adapters: I/O adapters — the graph’s edges to the outside world, built strictly on the public Op-pattern API (sources over channel / poll, sinks over for_each).
channel: The channel layer: cross-thread / cross-process value transport with a typed Message envelope, ported onto the Op model (Phase 3).
fluent: Fluent wiring sugar: the legacy wingfoil chaining style (ticker(d).count().map(f).filter(&cond)) over the explicit Builder core.
interp: The interpreted engine: dynamic wiring and execution of Ops.
introspect: Static graph introspection: the wired topology as data, and as pictures.
latency: Latency capture for wingfoil — stamp wall-clock timestamps onto messages as they hop through ops (and across processes), then aggregate the per-stage deltas at the end of the pipeline.
op: The Op trait: node semantics as pure, monomorphizable functions.
ops: The core op vocabulary. Each op is a zero-sized witness type carrying semantics in associated functions — never instantiated. Compare each cycle body with the corresponding MutableNode impl in the main crate: the logic is identical, but here it is written once and executed by every engine, interpreted or compiled.
pool: Pooled payloads: recycled buffers behind cheap graph-side handles, and the loan-based channel producer API (pooled_channel).
prelude: The common wiring vocabulary, re-exported for use wingfoil::prelude::*.
runtime: The shared runtime core: engine time, run bounds, the scheduled-callback queue, the burst grouping type and the Kernel that drives a run.
tier: Tier: which of a nitro! module’s engines executes the graph.

Macros§

__wf_fluent_accumulate: Fluent wiring for [accumulate], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_accumulate!(T); — to define the accumulate method over the trait’s element type T.
__wf_fluent_buffer: Fluent wiring for [buffer], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_buffer!(T); — to define the buffer method over the trait’s element type T.
__wf_fluent_collapse: Fluent wiring for [collapse], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_collapse!(T); — to define the collapse method over the trait’s element type T.
__wf_fluent_constant: Fluent wiring for [constant], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_constant!(); — to define the constant method on a GraphBuilder extension trait (it is a source — no receiver stream).
__wf_fluent_cumulative_max: Fluent wiring for [cumulative_max], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_max!(); — to define the cumulative_max method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_mean: Fluent wiring for [cumulative_mean], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_mean!(); — to define the cumulative_mean method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_mean_time_weighted: Fluent wiring for [cumulative_mean_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_mean_time_weighted!(); — to define the cumulative_mean_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_median: Fluent wiring for [cumulative_median], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_median!(); — to define the cumulative_median method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_median_time_weighted: Fluent wiring for [cumulative_median_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_median_time_weighted!(); — to define the cumulative_median_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_min: Fluent wiring for [cumulative_min], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_min!(); — to define the cumulative_min method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_std: Fluent wiring for [cumulative_std], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_std!(); — to define the cumulative_std method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_std_time_weighted: Fluent wiring for [cumulative_std_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_std_time_weighted!(); — to define the cumulative_std_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_sum: Fluent wiring for [cumulative_sum], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_sum!(); — to define the cumulative_sum method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_var: Fluent wiring for [cumulative_var], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_var!(); — to define the cumulative_var method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_cumulative_var_time_weighted: Fluent wiring for [cumulative_var_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_cumulative_var_time_weighted!(); — to define the cumulative_var_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_delay: Fluent wiring for [delay], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_delay!(T); — to define the delay method over the trait’s element type T.
__wf_fluent_difference: Fluent wiring for [difference], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_difference!(T); — to define the difference method over the trait’s element type T.
__wf_fluent_distinct: Fluent wiring for [distinct], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_distinct!(T); — to define the distinct method over the trait’s element type T.
__wf_fluent_drop_small_change: Fluent wiring for [drop_small_change], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_drop_small_change!(T); — to define the drop_small_change method over the trait’s element type T.
__wf_fluent_enumerate: Fluent wiring for [enumerate], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_enumerate!(T); — to define the enumerate method over the trait’s element type T.
__wf_fluent_ewma: Fluent wiring for [ewma], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_ewma!(); — to define the ewma method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_ewma_half_life: Fluent wiring for [ewma_half_life], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_ewma_half_life!(); — to define the ewma_half_life method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_filter: Fluent wiring for [filter], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_filter!(T); — to define the filter method over the trait’s element type T.
__wf_fluent_filter_value: Fluent wiring for [filter_value], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_filter_value!(T); — to define the filter_value method over the trait’s element type T.
__wf_fluent_finally: Fluent wiring for [finally], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_finally!(T); — to define the finally method over the trait’s element type T.
__wf_fluent_fold: Fluent wiring for [fold], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_fold!(T); — to define the fold method over the trait’s element type T.
__wf_fluent_for_each: Fluent wiring for [for_each], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_for_each!(T); — to define the for_each method over the trait’s element type T.
__wf_fluent_inspect: Fluent wiring for [inspect], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_inspect!(T); — to define the inspect method over the trait’s element type T.
__wf_fluent_join: Fluent wiring for [join], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_join!(T); — to define the join method over the trait’s element type T.
__wf_fluent_join3: Fluent wiring for [join3], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_join3!(T); — to define the join3 method over the trait’s element type T.
__wf_fluent_join_passive: Fluent wiring for [join_passive], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_join_passive!(T); — to define the join_passive method over the trait’s element type T.
__wf_fluent_limit: Fluent wiring for [limit], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_limit!(T); — to define the limit method over the trait’s element type T.
__wf_fluent_map: Fluent wiring for [map], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_map!(T); — to define the map method over the trait’s element type T.
__wf_fluent_map_filter: Fluent wiring for [map_filter], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_map_filter!(T); — to define the map_filter method over the trait’s element type T.
__wf_fluent_merge: Fluent wiring for [merge], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_merge!(T); — to define the merge method over the trait’s element type T.
__wf_fluent_not: Fluent wiring for [not], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_not!(T); — to define the not method over the trait’s element type T.
__wf_fluent_pairwise: Fluent wiring for [pairwise], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_pairwise!(T); — to define the pairwise method over the trait’s element type T.
__wf_fluent_print: Fluent wiring for print, generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_print!(T); — to define the print method over the trait’s element type T.
__wf_fluent_rolling_max: Fluent wiring for [rolling_max], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_max!(); — to define the rolling_max method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_mean: Fluent wiring for [rolling_mean], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_mean!(); — to define the rolling_mean method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_mean_time_weighted: Fluent wiring for [rolling_mean_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_mean_time_weighted!(); — to define the rolling_mean_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_median: Fluent wiring for [rolling_median], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_median!(); — to define the rolling_median method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_median_time_weighted: Fluent wiring for [rolling_median_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_median_time_weighted!(); — to define the rolling_median_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_min: Fluent wiring for [rolling_min], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_min!(); — to define the rolling_min method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_std: Fluent wiring for [rolling_std], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_std!(); — to define the rolling_std method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_std_time_weighted: Fluent wiring for [rolling_std_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_std_time_weighted!(); — to define the rolling_std_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_sum: Fluent wiring for [rolling_sum], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_sum!(); — to define the rolling_sum method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_var: Fluent wiring for [rolling_var], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_var!(); — to define the rolling_var method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_rolling_var_time_weighted: Fluent wiring for [rolling_var_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_rolling_var_time_weighted!(); — to define the rolling_var_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_sample: Fluent wiring for [sample], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_sample!(T); — to define the sample method over the trait’s element type T.
__wf_fluent_scan: Fluent wiring for [scan], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_scan!(T); — to define the scan method over the trait’s element type T.
__wf_fluent_skip: Fluent wiring for [skip], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_skip!(T); — to define the skip method over the trait’s element type T.
__wf_fluent_skip_while: Fluent wiring for [skip_while], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_skip_while!(T); — to define the skip_while method over the trait’s element type T.
__wf_fluent_step_by: Fluent wiring for [step_by], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_step_by!(T); — to define the step_by method over the trait’s element type T.
__wf_fluent_take_while: Fluent wiring for [take_while], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_take_while!(T); — to define the take_while method over the trait’s element type T.
__wf_fluent_throttle: Fluent wiring for [throttle], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_throttle!(T); — to define the throttle method over the trait’s element type T.
__wf_fluent_ticked_at: Fluent wiring for [ticked_at], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_ticked_at!(T); — to define the ticked_at method over the trait’s element type T.
__wf_fluent_ticked_at_elapsed: Fluent wiring for [ticked_at_elapsed], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_ticked_at_elapsed!(T); — to define the ticked_at_elapsed method over the trait’s element type T.
__wf_fluent_ticker: Fluent wiring for [ticker], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_ticker!(); — to define the ticker method on a GraphBuilder extension trait (it is a source — no receiver stream).
__wf_fluent_time_windowed_max: Fluent wiring for [time_windowed_max], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_max!(); — to define the time_windowed_max method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_mean: Fluent wiring for [time_windowed_mean], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_mean!(); — to define the time_windowed_mean method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_mean_time_weighted: Fluent wiring for [time_windowed_mean_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_mean_time_weighted!(); — to define the time_windowed_mean_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_median: Fluent wiring for [time_windowed_median], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_median!(); — to define the time_windowed_median method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_median_time_weighted: Fluent wiring for [time_windowed_median_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_median_time_weighted!(); — to define the time_windowed_median_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_min: Fluent wiring for [time_windowed_min], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_min!(); — to define the time_windowed_min method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_std: Fluent wiring for [time_windowed_std], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_std!(); — to define the time_windowed_std method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_std_time_weighted: Fluent wiring for [time_windowed_std_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_std_time_weighted!(); — to define the time_windowed_std_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_sum: Fluent wiring for [time_windowed_sum], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_sum!(); — to define the time_windowed_sum method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_var: Fluent wiring for [time_windowed_var], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_var!(); — to define the time_windowed_var method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_time_windowed_var_time_weighted: Fluent wiring for [time_windowed_var_time_weighted], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_time_windowed_var_time_weighted!(); — to define the time_windowed_var_time_weighted method over Stream<f64>, the concrete receiver its first edge fixes.
__wf_fluent_timed: Fluent wiring for [timed], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_timed!(T); — to define the timed method over the trait’s element type T.
__wf_fluent_try_join: Fluent wiring for [try_join], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_try_join!(T); — to define the try_join method over the trait’s element type T.
__wf_fluent_try_join3: Fluent wiring for [try_join3], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_try_join3!(T); — to define the try_join3 method over the trait’s element type T.
__wf_fluent_try_join_passive: Fluent wiring for [try_join_passive], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_try_join_passive!(T); — to define the try_join_passive method over the trait’s element type T.
__wf_fluent_try_map: Fluent wiring for [try_map], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_try_map!(T); — to define the try_map method over the trait’s element type T.
__wf_fluent_try_map_filter: Fluent wiring for [try_map_filter], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_try_map_filter!(T); — to define the try_map_filter method over the trait’s element type T.
__wf_fluent_window: Fluent wiring for [window], generated by #[op(fluent)]. Invoke it inside an extension-trait impl — __wf_fluent_window!(T); — to define the window method over the trait’s element type T.
burst: Macro to create a Burst<T> with type inference.
nitro: One wiring definition, two engines: expands to a module with interpreted() (fluent wiring), compiled(run_mode, run_for) (fully monomorphized runner) and run(tier, ..) (either, same outputs) emitted from the same tokens. See wingfoil_derive for the DSL. See the crate docs: fluent wiring in, interpreted() + compiled() out.

Structs§

Kernel: The shared runtime core, re-exported at the crate root: engine time, the run bounds, the scheduled-callback queue and the Kernel that drives a run. The legacy wingfoil crate re-exports these same items — they are one set of types, not two — see runtime for why the core lives here. Clock, scheduled-callback queue and run bounds for a kernel-driven engine. See the module docs for how this relates to the interpreted engine.
KernelWaker: The shared runtime core, re-exported at the crate root: engine time, the run bounds, the scheduled-callback queue and the Kernel that drives a run. The legacy wingfoil crate re-exports these same items — they are one set of types, not two — see runtime for why the core lives here. Wakes a realtime Kernel from another thread, marking a node dirty — the kernel-level equivalent of the interpreted engine’s ReadyNotifier. Cheap to clone; hand one to each producer thread / async task.
NanoTime: A time in nanoseconds since the unix epoch.
TimeQueue: A time-ordered queue of Ts, earliest first.

Enums§

RunFor: Defines how long the graph should run for. Can be a Duration, number of cycles or forever.
RunMode: Whether the graph should run in RealTime or Historical mode.
Tier: Which nitro! engine executes a graph, for the generated run(tier, run_mode, run_for) — interpreted while developing, compiled in production, one argument apart. Which of a nitro! module’s engines executes the graph.
TimerPolicy: The shared runtime core, re-exported at the crate root: engine time, the run bounds, the scheduled-callback queue and the Kernel that drives a run. The legacy wingfoil crate re-exports these same items — they are one set of types, not two — see runtime for why the core lives here. How a realtime kernel waits out the gap to the next scheduled callback.

Functions§

waker_channel: The shared runtime core, re-exported at the crate root: engine time, the run bounds, the scheduled-callback queue and the Kernel that drives a run. The legacy wingfoil crate re-exports these same items — they are one set of types, not two — see runtime for why the core lives here. Create a waker/receiver pair for external (threaded/async) sources. Hand the KernelWaker to producers and the receiver to Kernel::with_ready.

Type Aliases§

Burst: A group of same-instant values, delivered atomically in one cycle (never coalesced / latest-wins). A tinyvec::TinyVec<[T; 1]>, defined here and re-exported by the legacy wingfoil crate (with its burst! constructor macro), so both engines share one grouping type. A small vector optimised for single-element bursts.
ReadyReceiver: The shared runtime core, re-exported at the crate root: engine time, the run bounds, the scheduled-callback queue and the Kernel that drives a run. The legacy wingfoil crate re-exports these same items — they are one set of types, not two — see runtime for why the core lives here. The receiving half of a waker_channel, to be handed to Kernel::with_ready.

Attribute Macros§

op: Turn an impl Op for … block into a first-class op: the interpreted wiring (an extension trait on interp::Builder), the naming-convention forwarders nitro!’s compiled() / nested() emission dispatches through, and — with fluent — the macro_rules! that writes the op’s fluent method.

Crate wingfoil

Crate wingfoil Copy item path

§The idea: node semantics as a function, not an object

§The tour

§Threading: a graph lives on one thread

§Crossing the thread boundary

§Tracing and instrumentation

§Known limits

Re-exports§

Modules§

Macros§

Structs§

Enums§

Functions§

Type Aliases§

Attribute Macros§

Crate wingfoil