Pub/sub fan-out & dispatch cost (L4 graph)

One publisher, a growing set of subscribers, and a question: what does each extra subscriber cost? This example registers one /sensor/temp vertex, then for each fan-out width in {1, 8, 64} attaches that many callbacks, writes 5,000 values, and reports the per-delivery latency and delivery throughput. Every write fans out to every subscriber as a refcount-bump clone of the same rope value — no byte copy per subscriber.

What to notice

  • Fan-out is amortized, not multiplied — per-delivery cost drops as fan-out grows (the per-write overhead is shared across more subscribers); the RESULT lines make the trend visible.

  • The value is cloned by refcount, not by bytes — 64 subscribers means 64 refcount bumps on one segment, not 64 copies.

  • :schema discovery — the vertex’s shape is read back structurally as a 2-child POINT (NAME plus the synthesized, and since RFC-0022 §3.B empty, SETTINGS).

  • The RESULT line is informational — timing never fails CI; the self-checks assert every subscriber saw every write.

Note

The absolute nanoseconds come from whatever build ran (CI builds the examples in a debug configuration); treat them as a shape, not a spec number. The canonical, release-build, CI-published 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 Pub/sub fan-out — one publisher, a growing set of subscribers, and the
  9 *        per-delivery dispatch cost as fan-out scales.
 10 *
 11 * Every `write` fans out to every subscriber as a refcount-bump clone of the same
 12 * rope value — no byte copy per subscriber (`docs/modules/graph.md`). This example
 13 * registers one `/sensor/temp` vertex, then for each fan-out width in {1, 8, 64}
 14 * attaches that many callbacks, writes @p kWrites values, and reports the
 15 * per-delivery latency and delivery throughput. It also field-writes a QoS
 16 * setting and reads it back structurally via `:schema`.
 17 *
 18 * The RESULT perf lines are informational (CI never flakes on timing); the
 19 * self-checks guard that EVERY subscriber saw EVERY write. Runs under ctest as
 20 * `example_pubsub_fanout`.
 21 */
 22
 23#include <chrono>
 24#include <cstddef>
 25#include <cstdint>
 26#include <cstdio>
 27#include <vector>
 28
 29#include "libtracer/tracer.hpp"
 30
 31namespace {
 32
 33using clock_t_ = std::chrono::steady_clock;
 34using tr::graph::path_t;
 35using tr::graph::role_t;
 36
 37/** @brief A 4-byte little-endian VALUE view over @p v (one heap segment). */
 38tr::view::view_t value_u32(std::uint32_t v) {
 39    tr::view::segment_ptr_t seg = tr::view::heap_alloc(4);
 40    for (int i = 0; i < 4; ++i)
 41        seg->bytes[static_cast<std::size_t>(i)] = static_cast<std::byte>((v >> (8 * i)) & 0xFF);
 42    return tr::view::view_t::over(std::move(seg));
 43}
 44
 45/** @brief Record a failed expectation on @p ok and report it. */
 46void check(bool& ok, bool cond, const char* what) {
 47    if (!cond) {
 48        std::printf("  [FAIL] %s\n", what);
 49        ok = false;
 50    }
 51}
 52
 53/** @brief Run one fan-out width: @p fanout subscribers, @p writes writes; returns ok. */
 54bool run_fanout(std::size_t fanout, std::uint32_t writes, bool& ok) {
 55    tr::graph::graph_t g;
 56    const tr::graph::vertex_handle_t v =
 57        g.register_vertex(path_t("/sensor/temp"), role_t::STORED_VALUE);
 58
 59    // One delivery counter shared by every callback — each fires inline per write.
 60    std::uint64_t deliveries = 0;
 61    auto on_temp = [&deliveries](const tr::view::rope_t&) { ++deliveries; };
 62    for (std::size_t s = 0; s < fanout; ++s) (void)g.subscribe(path_t("/sensor/temp"), on_temp);
 63
 64    auto t0 = clock_t_::now();
 65    for (std::uint32_t i = 0; i < writes; ++i) (void)g.write(v, value_u32(i));
 66    auto t1 = clock_t_::now();
 67
 68    const std::uint64_t expected = std::uint64_t(fanout) * writes;
 69    check(ok, deliveries == expected, "every subscriber received every write");
 70
 71    const double total_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
 72    const double per_delivery_ns = total_ns / double(expected);
 73    const double deliv_per_s = double(expected) / (total_ns * 1e-9);
 74    std::printf(
 75        "RESULT pubsub_fanout subs=%zu writes=%u deliveries=%llu "
 76        "per_delivery_ns=%.1f deliveries_Mps=%.2f\n",
 77        fanout, writes, static_cast<unsigned long long>(deliveries), per_delivery_ns,
 78        deliv_per_s / 1e6);
 79    return ok;
 80}
 81
 82}  // namespace
 83
 84int main() {
 85    constexpr std::uint32_t kWrites = 5000;
 86    bool ok = true;
 87
 88    std::printf(
 89        "fan-out dispatch: each write clones the rope per subscriber "
 90        "(refcount bump, no byte copy)\n");
 91    for (std::size_t fanout : {std::size_t{1}, std::size_t{8}, std::size_t{64}})
 92        run_fanout(fanout, kWrites, ok);
 93
 94    // Declare the STREAM ring depth OWNER-SIDE (RFC-0022 §3.C — it has no wire surface),
 95    // then discover the vertex shape via :schema.
 96    tr::graph::graph_t g;
 97    const auto temp = g.register_vertex(path_t("/sensor/temp"), role_t::STORED_VALUE);
 98    g.set_history_depth(temp, 8);
 99    auto schema = g.read(path_t("/sensor/temp:schema"));
100    std::size_t schema_children = 0;
101    if (schema)
102        if (auto point = tr::wire::decode((*schema)->only()))
103            schema_children = point->children.size();
104    std::printf(":schema resolves to a POINT with %zu children\n", schema_children);
105    check(ok, schema_children == 2, ":schema is a 2-child POINT");
106
107    std::printf("%s\n", ok ? "pub/sub fan-out OK" : "pub/sub fan-out FAILED");
108    return ok ? 0 : 1;
109}

See also: graph module · in-process pub/sub · performance.