Two nodes over a wire — FWD delivery (L4 + transport)

Two independent graph_t nodes, each with a fwd_router_t, connected by a loopback_channel_t — the in-process dev “wire”. A client hands node A’s router an FWD{ op=WRITE, dst=/b/sensor/temp } frame; A peeks the first dst segment, strips b, and forwards /sensor/temp across the wire; B’s terminus writes it into B’s local vertex, waking B’s subscriber. The route is explicit and loop-free by construction — no per-request state on any hop (RFC-0004, ADR-0040; network formation reference).

Tip

Going to a real socket is a one-line swap. Replace channel.a() / channel.b() with two tr::net::udp_transport_t instances and the routers/graphs above are unchanged — that exact swap is core/tests/udp_test.cpp’s two-node test.

What to notice

  • The frame carries its own route — fwd_write builds FWD{ op, dst, src, payload }; dst shrinks one NAME per hop and src grows the way back, so the reply route is assembled for free.

  • The forward hop never touches the heap — a router reads three headers by offset and scatter-gathers the shrunk-dst / grown-src heads with untouched views of the inbound frame.

  • Cross-wire latency is measured — 2,000 sequential FWD writes through the loopback channel; the RESULT line reports the mean end-to-end delivery time (dispatch + the channel’s receive-thread hand-off).

  • It self-checks — every frame must be delivered with the exact payload, so the ctest smoke test guards delivery, not just a clean exit.

Note

The absolute nanoseconds come from whatever build ran (CI builds the examples in a debug configuration) and include the loopback channel’s thread hand-off; treat them as a shape, not a spec number. The canonical network latency/throughput figures live on the performance page.

Source

  1/*
  2 * SPDX-License-Identifier: Apache-2.0
  3 * SPDX-FileCopyrightText: Copyright 2026 avatarsd LLC
  4 */
  5
  6/**
  7 * @file
  8 * @brief Two nodes over a wire — an FWD write routed between two graphs, and the
  9 *        end-to-end delivery latency across the "wire".
 10 *
 11 * Two independent `graph_t` nodes, each with a `fwd_router_t`, connected by a
 12 * `loopback_channel_t` — the in-process dev "wire" (`docs/reference/13-network-formation.md`,
 13 * ADR-0040). A client hands node A's router an `FWD{ op=WRITE, dst=/b/sensor/temp }`
 14 * frame; A peeks the first `dst` segment, strips `b`, and forwards `/sensor/temp`
 15 * across the wire; B's terminus writes it into B's local vertex, waking B's
 16 * subscriber. The route is explicit and loop-free by construction — no per-request
 17 * state on any hop.
 18 *
 19 * Going to a real socket is a one-line swap: replace `channel.a()`/`channel.b()`
 20 * with two `udp_transport_t` — exactly `core/tests/udp_test.cpp`'s two-node test.
 21 *
 22 * The RESULT perf line (mean cross-wire delivery latency) is informational so CI
 23 * never flakes on timing; the self-checks guard that every frame is delivered with
 24 * the exact payload. Runs under ctest as `example_two_node_fwd`.
 25 */
 26
 27#include <chrono>
 28#include <condition_variable>
 29#include <cstddef>
 30#include <cstdint>
 31#include <cstdio>
 32#include <mutex>
 33#include <span>
 34#include <string_view>
 35#include <vector>
 36
 37#include "libtracer/fwd_router.hpp"
 38#include "libtracer/loopback.hpp"
 39#include "libtracer/tlv_emit.hpp"
 40#include "libtracer/tracer.hpp"
 41
 42namespace {
 43
 44using namespace std::chrono_literals;
 45using clock_t_ = std::chrono::steady_clock;
 46using tr::graph::graph_t;
 47using tr::graph::path_t;
 48using tr::graph::role_t;
 49
 50/** @brief A VALUE TLV (encoded wire bytes) carrying @p bytes. */
 51std::vector<std::byte> value_tlv(std::initializer_list<std::uint8_t> bytes) {
 52    std::vector<std::byte> payload;
 53    for (std::uint8_t b : bytes) payload.push_back(std::byte{b});
 54    tr::wire::tlv_t t{.type = tr::wire::type_t::VALUE, .payload = payload};
 55    return tr::wire::encode(t);
 56}
 57
 58/**
 59 * @brief Build FWD{ op=WRITE, dst=<segs…>, src=<empty>, payload=<VALUE> } — a remote
 60 *        write routed by explicit source route (RFC-0004 §D, ADR-0040).
 61 */
 62std::vector<std::byte> fwd_write(std::initializer_list<std::string_view> dst,
 63                                 std::span<const std::byte> payload_value_tlv) {
 64    std::vector<std::byte> body;
 65    const std::byte op{static_cast<std::uint8_t>(tr::graph::fwd_op_t::WRITE)};
 66    tr::wire::emit_tlv(body, tr::wire::type_t::VALUE, tr::wire::opt_t{},
 67                       std::span<const std::byte>(&op, 1));
 68    std::vector<std::byte> dst_segs;
 69    for (std::string_view s : dst) tr::wire::emit_name(dst_segs, s);
 70    tr::wire::emit_tlv(body, tr::wire::type_t::PATH, tr::wire::opt_t{.pl = true}, dst_segs);
 71    tr::wire::emit_tlv(body, tr::wire::type_t::PATH, tr::wire::opt_t{.pl = true},
 72                       std::span<const std::byte>{});  // src: empty, grows per hop
 73    body.insert(body.end(), payload_value_tlv.begin(), payload_value_tlv.end());
 74    std::vector<std::byte> frame;
 75    tr::wire::emit_tlv(frame, tr::wire::type_t::FWD, tr::wire::opt_t{.pl = true}, body);
 76    return frame;
 77}
 78
 79}  // namespace
 80
 81int main() {
 82    // Declaration order: the channel is declared LAST so it destructs FIRST — its
 83    // receive threads join before the routers they call into are gone.
 84    graph_t node_a, node_b;
 85    tr::net::fwd_router_t router_a(node_a);
 86    tr::net::fwd_router_t router_b(node_b);
 87    tr::net::loopback_channel_t channel;
 88
 89    // B owns the target vertex; A knows its link to B as "b", B knows its link back as "a".
 90    (void)node_b.register_vertex(path_t("/sensor/temp"), role_t::STORED_VALUE);
 91    router_a.add_child("b", channel.a());  // a `dst` starting with "b" routes over the wire
 92    router_b.add_child("a", channel.b());  // B's name for the inbound link (src accumulation)
 93
 94    // B's subscriber signals each delivery; we count them and keep the last payload.
 95    std::mutex m;
 96    std::condition_variable cv;
 97    std::uint64_t delivered = 0;
 98    std::vector<std::byte> last;
 99    (void)node_b.subscribe(path_t("/sensor/temp"), [&](const tr::view::rope_t& v) {
100        const auto b = v.only().bytes();
101        std::lock_guard<std::mutex> lk(m);
102        last.assign(b.begin(), b.end());
103        ++delivered;
104        cv.notify_one();
105    });
106
107    bool ok = true;
108    const auto payload = value_tlv({0x2A, 0x2B});
109
110    // One delivery for correctness: A routes /b/sensor/temp → strips "b" → B writes it.
111    router_a.on_frame("client", fwd_write({"b", "sensor", "temp"}, payload));
112    {
113        std::unique_lock<std::mutex> lk(m);
114        const bool arrived = cv.wait_for(lk, 3s, [&] { return delivered >= 1; });
115        if (!arrived) {
116            std::printf("  [FAIL] node B never received the FWD write\n");
117            ok = false;
118        } else if (last != payload) {
119            std::printf("  [FAIL] delivered payload differs from what A sent\n");
120            ok = false;
121        } else {
122            std::printf(
123                "node B received the FWD-delivered value across the wire "
124                "(explicit source route /b/sensor/temp → /sensor/temp)\n");
125        }
126    }
127
128    // --- perf: mean cross-wire delivery latency over many sequential FWD writes ---
129    constexpr std::uint64_t kMsgs = 2000;
130    const std::uint64_t base = delivered;
131    auto t0 = clock_t_::now();
132    for (std::uint64_t i = 0; i < kMsgs; ++i)
133        router_a.on_frame("client", fwd_write({"b", "sensor", "temp"}, payload));
134    std::uint64_t target = base + kMsgs;
135    {
136        std::unique_lock<std::mutex> lk(m);
137        const bool all = cv.wait_for(lk, 10s, [&] { return delivered >= target; });
138        if (!all) {
139            std::printf("  [FAIL] only %llu/%llu frames delivered\n",
140                        static_cast<unsigned long long>(delivered - base),
141                        static_cast<unsigned long long>(kMsgs));
142            ok = false;
143        }
144    }
145    auto t1 = clock_t_::now();
146
147    const double total_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
148    const double per_ns = total_ns / double(kMsgs);
149    std::printf("RESULT two_node_fwd msgs=%llu mean_delivery_ns=%.0f throughput_Kps=%.1f\n",
150                static_cast<unsigned long long>(kMsgs), per_ns,
151                (double(kMsgs) / (total_ns * 1e-9)) / 1e3);
152    std::printf(
153        "each hop reads three headers by offset and scatter-gathers the shrunk-dst / "
154        "grown-src heads — the forward hop never touches the heap.\n");
155
156    std::printf("%s\n", ok ? "two-node FWD OK" : "two-node FWD FAILED");
157    return ok ? 0 : 1;
158}

See also: transport module · network formation reference · communication flows.