Networks¶

A Network wires nodes together at runtime using a builder chain.

Building a network¶

    Network net;
    net.add("src",  src)
       .add("dbl",  dbl)
       .add("sink", sink)
       .connect("src",  src.output<0>(),  "dbl",  dbl.input<0>())
       .connect("dbl",  dbl.output<0>(),  "sink", sink.input<0>())
       .build();

    net.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    net.stop();

The builder chain:

Method	Purpose
`.add(name, node)`	Register a node; assigns its name
`.connect(src, port, dst, port)`	Wire one output port to one input port
`.build()`	Compute topological order; inject network callbacks
`.start()`	Start nodes in topological order
`.stop()`	Stop all nodes immediately
`.shutdown()`	Graceful drain: stop sources first, wait for channels to empty, then stop downstream

Port access¶

Ports are accessed by index or by name:

// By index
net.connect("src", src.output<0>(), "dst", dst.input<0>());

// By name (requires named ports)
    Network net;
    net.add("tok", tok)
       .add("cnt", cnt)
       .add("snk", snk)
       .connect("tok", tok.template output<"words">(), "cnt", cnt.template input<"words">())
       .connect("cnt", cnt.template output<"count">(), "snk", snk.template input<"count">())
       .connect("cnt", cnt.template output<"words">(), "snk", snk.template input<"words">())
       .build();

    net.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    net.stop();

Diagnostics¶

Install a diagnostics handler to receive periodic snapshots of every node and channel:

    // Custom diagnostics handler — fires on the watchdog interval.
    // Print a concise one-liner rather than the full table.
    net.set_diagnostics_handler([](const std::vector<NodeSnapshot>& nodes,
                                   const std::vector<ChannelSnapshot>& channels) {
        std::cout << "[diag] ";
        for (auto& n : nodes)
            std::cout << n.name << "=" << n.throughput_fps << "fps  ";
        for (auto& c : channels)
            std::cout << "channel fill=" << static_cast<int>(c.fill_pct()) << "% "
                      << "overflows=" << c.overflows;
        std::cout << '\n';
    });

Or print a full report at any time:

net.print_diagnostics();   // writes to stderr by default
net.print_diagnostics(std::cout);

Network-level event handler¶

Observe overflow and node-stop events across the entire network in one place:

    // Network-level aggregate handler — covers every node, includes node name.
    net.set_event_handler([](std::string_view name, NodeEvent ev,
                             steady_clock::time_point ts) {
        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
        std::string_view kind = (ev == NodeEvent::Overflow) ? "overflow" : "closed";
        std::cerr << "[net:" << kind << "] node=" << name << " t=" << ms << "ms\n";
    });

NodeEvent is either NodeEvent::Overflow (item dropped on full channel) or NodeEvent::Closed (node stopped due to crash or closed upstream channel). See Error Handling & Events.

Shutdown¶

net.stop() halts immediately — all nodes stop in reverse topological order.

net.shutdown() drains gracefully: source nodes stop first; their output channels are polled until empty; then the next layer stops, and so on. This ensures no items are lost if downstream nodes are still consuming.

StaticNetwork¶

For zero-overhead compile-time topology, see Static Networks.