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 — after a :settings.deadline_ns field-write, the vertex’s shape is read back structurally as a 2-child POINT.

  • 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    for (std::size_t s = 0; s < fanout; ++s)
 62        (void)g.subscribe(path_t("/sensor/temp"),
 63                          [&deliveries](const tr::view::rope_t&) { ++deliveries; });
 64
 65    auto t0 = clock_t_::now();
 66    for (std::uint32_t i = 0; i < writes; ++i) (void)g.write(v, value_u32(i));
 67    auto t1 = clock_t_::now();
 68
 69    const std::uint64_t expected = std::uint64_t(fanout) * writes;
 70    check(ok, deliveries == expected, "every subscriber received every write");
 71
 72    const double total_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
 73    const double per_delivery_ns = total_ns / double(expected);
 74    const double deliv_per_s = double(expected) / (total_ns * 1e-9);
 75    std::printf(
 76        "RESULT pubsub_fanout subs=%zu writes=%u deliveries=%llu "
 77        "per_delivery_ns=%.1f deliveries_Mps=%.2f\n",
 78        fanout, writes, static_cast<unsigned long long>(deliveries), per_delivery_ns,
 79        deliv_per_s / 1e6);
 80    return ok;
 81}
 82
 83}  // namespace
 84
 85int main() {
 86    constexpr std::uint32_t kWrites = 5000;
 87    bool ok = true;
 88
 89    std::printf(
 90        "fan-out dispatch: each write clones the rope per subscriber "
 91        "(refcount bump, no byte copy)\n");
 92    for (std::size_t fanout : {std::size_t{1}, std::size_t{8}, std::size_t{64}})
 93        run_fanout(fanout, kWrites, ok);
 94
 95    // Field-write a QoS setting, then discover the vertex shape via :schema.
 96    tr::graph::graph_t g;
 97    (void)g.register_vertex(path_t("/sensor/temp"), role_t::STORED_VALUE);
 98    (void)g.write(path_t("/sensor/temp:settings.deadline_ns"), value_u32(5000));
 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::view_as_tlv(schema->only()))
103            schema_children = point->children.size();
104    std::printf(":schema resolves to a POINT with %zu children after the field-write\n",
105                schema_children);
106    check(ok, schema_children == 2, ":schema is a 2-child POINT");
107
108    std::printf("%s\n", ok ? "pub/sub fan-out OK" : "pub/sub fan-out FAILED");
109    return ok ? 0 : 1;
110}

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