From 5cc7b2de2a1bb0d935ee53257e23019bb55a85e9 Mon Sep 17 00:00:00 2001 From: Joe Rivera Date: Thu, 22 Oct 2026 00:47:16 -0500 Subject: [PATCH 1/3] =?UTF-8?q?feat(core):=20clock=5Fdriver=20=E2=80=94=20?= =?UTF-8?q?one-advance-per-frame=20fan-out=20over=20world=5Fclock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce cvc::clock_driver (inc/cvc/core/clock_driver.h, src/cvc/core/clock_driver.cpp): the host-side loop that advances a world_clock exactly once per frame and fans the result out to ordered observers that share a single clock_frame. Two observer kinds mirror the advance()/for-each-step/present shape the world_clock header sketches and that tick_scene already open-codes: - step observers fire first, ascending `order`: "advance your own sim by the quanta that fell due" (physics follow, a holon's simulation). - frame observers fire after every step observer, ascending `order`: "do the once-per-frame work / present at frame.alpha" (drain a transport, forward media stamped with frame.t, interpolate a pose). Every observer sees the SAME clock_frame computed once at the top of advance(), so a media timestamp (frame.t) read by one observer is bit-identical to the one read by another — what lets a receiver align two streams stamped in the same frame. clock_frame also carries the raw wall_dt for work that must track real time regardless of sim rate (talk-hold decay, camera, fps), a monotonic frame index, and a `first` flag; a measured first advance books no startup burst. Optional world_clock_binding routing lets the sim.* control keys steer the driver and publishes telemetry each frame. Observers may be removed or added from inside the fan-out (a self-removing one-shot, deferred adds); re-entering advance() throws. This is the reusable spine for the holon tick model and for refactoring the conference host's monolithic per-frame pump onto an ordered, single-advance observer set. 13 gtest cases in src/cvc/tests/clock_driver_test.cpp cover ordering, the shared frame, first-frame seeding, remove/add during fan-out, re-entrancy, the follower-driven step path, and the binding route. --- inc/cvc/core/clock_driver.h | 174 ++++++++++++++++++ src/cvc/CMakeLists.txt | 3 + src/cvc/core/clock_driver.cpp | 160 +++++++++++++++++ src/cvc/tests/CMakeLists.txt | 11 ++ src/cvc/tests/clock_driver_test.cpp | 262 ++++++++++++++++++++++++++++ 5 files changed, 610 insertions(+) create mode 100644 inc/cvc/core/clock_driver.h create mode 100644 src/cvc/core/clock_driver.cpp create mode 100644 src/cvc/tests/clock_driver_test.cpp diff --git a/inc/cvc/core/clock_driver.h b/inc/cvc/core/clock_driver.h new file mode 100644 index 000000000..d07f4c4d8 --- /dev/null +++ b/inc/cvc/core/clock_driver.h @@ -0,0 +1,174 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. +*/ + +#ifndef __CVC_CLOCK_DRIVER_H__ +#define __CVC_CLOCK_DRIVER_H__ + +#include +#include +#include +#include +#include +#include + +namespace cvc { + +class world_clock_binding; + +// --------------- +// cvc::clock_driver +// --------------- +// The one place a host advances a world_clock per frame and fans the result out +// to everything that follows it, in a fixed order, sharing a single reading of +// time. +// +// A world_clock says "one driver per clock; everything else follows" but leaves +// the driving to the host. In practice every real host grew the same loop by +// hand: measure a wall delta, call advance() exactly once, then -- in a +// load-bearing order -- step the simulations whose quanta fell due and present +// what the frame should show at the returned alpha. The Ariadne scene driver +// (tick_scene) is this loop; the conference host is this loop open-coded as an +// eleven-step pump; a holon will be this loop again. clock_driver is that loop +// factored out so there is exactly ONE advance per clock per frame and one place +// that owns the order. +// +// Two kinds of observer, mirroring the advance()/for-each-step/present shape the +// world_clock header sketches: +// +// step observers - fire first, in ascending `order`. Semantically "advance +// your own state by the quanta that fell due this frame" (a physics world's +// follow(), a navigator, a holon's simulation). They read frame.step +// (first_tick + 1 .. first_tick + steps) or keep a cvc::clock_follower; they +// must not advance the clock. +// frame observers - fire after every step observer, in ascending `order`. +// Semantically "do the once-per-frame work / present at frame.alpha" (drain +// a transport, forward media stamped with frame.t, interpolate a pose, +// publish telemetry). The conference's whole pump is frame observers. +// +// Both kinds receive the SAME clock_frame, computed once at the top of advance() +// -- so a media timestamp (frame.t) read by one observer is bit-identical to the +// one read by another, which is what lets a receiver align two streams that were +// stamped in the same frame. +// +// Threading: a clock_driver is single-threaded. advance() and the observers it +// calls run on whatever thread drives the frame (the SDL/render main thread for +// a window host). It takes no locks and must not be driven from two threads; a +// follower on another thread waits on the clock's wait_for_tick_change() and +// keeps its own cursor, it does not share a driver. Re-entering advance() from +// within an observer throws. +class clock_driver { +public: + // A frame of time, computed once per advance() and handed to every observer. + struct clock_frame { + // What this advance decided: steps (whole quanta to run now), alpha + // ([0,1) into the next quantum), dropped_steps, first_tick (the tick BEFORE + // this advance; the quanta to run are first_tick + 1 .. first_tick + steps). + world_clock::step_result step; + std::uint64_t tick = 0; // clock.tick() AFTER this advance (first_tick + steps) + double t = 0.0; // world seconds, clock.t() (tick * fixed_dt) + double alpha = 0.0; // step.alpha; interpolation fraction into the next quantum + double fixed_dt = 0.0; // the clock's quantum + // The raw wall delta this frame, in seconds, BEFORE any time scaling or + // pause -- for work that must track real time no matter the sim rate + // (a talk-hold decay, a camera, an fps readout). Never the scaled world dt. + double wall_dt = 0.0; + bool first = false; // true only on the very first advance (step.steps == 0) + std::uint64_t index = 0; // monotonic frame counter, 0 on the first advance + }; + + using observer_fn = std::function; + + // A registration token. Keep it to remove() the observer later; a default + // (id == 0) handle names nothing. + struct handle { + std::uint64_t id = 0; + explicit operator bool() const noexcept { return id != 0; } + bool operator==(const handle &o) const noexcept { return id == o.id; } + }; + + explicit clock_driver(world_clock &clock); + + world_clock &clock() noexcept { return _clock; } + const world_clock &clock() const noexcept { return _clock; } + + // Route advance() through a world_clock_binding instead of the clock directly, + // so the sim.* control keys (speed/paused/mode/step) steer this driver and the + // telemetry keys (tick/time/dropped) are published each frame. nullptr (the + // default) drives the clock raw -- used where the clock is a private media + // timebase with no state face. The binding is not owned. + void set_binding(world_clock_binding *b) noexcept { _binding = b; } + world_clock_binding *binding() const noexcept { return _binding; } + + // Register an observer. `order` sorts observers of the same kind ascending; + // ties keep registration order (stable). Returns a handle for remove(). + // Registering from inside an observer is allowed; the new observer does not + // fire until the next advance(). + handle add_step_observer(int order, observer_fn fn); + handle add_frame_observer(int order, observer_fn fn); + + // Remove a previously registered observer. Safe to call from within an + // observer, including an observer removing itself (a one-shot). Returns true + // if a live observer with that handle was found. + bool remove(handle h); + + // Advance the clock once and fire observers: every step observer in order, + // then every frame observer in order, all with the same clock_frame. + // + // wall_dt < 0 (the default) measures a steady_clock delta since the previous + // advance -- real-time animation that does not crawl under a slow frame. The + // FIRST such call books no time (delta 0) and only seeds the baseline, so a + // host that constructs the driver long before its first frame does not inject + // a huge startup delta. wall_dt >= 0 uses the given delta verbatim (a fixed dt + // for deterministic capture, or a test), on the first call too. + // + // Returns the frame it built; the same reference frame() returns until the + // next advance. + const clock_frame &advance(double wall_dt = -1.0); + + const clock_frame &frame() const noexcept { return _frame; } + +private: + struct entry { + std::uint64_t id = 0; + int order = 0; + std::uint64_t seq = 0; // insertion order, breaks `order` ties stably + observer_fn fn; + bool dead = false; // removed while firing; compacted after the fan-out + bool want_step = false; // target list, used only for pending adds + }; + + handle add(std::vector &list, bool &dirty, int order, observer_fn fn); + static void sort_if_dirty(std::vector &list, bool &dirty); + void fire(std::vector &list); + void flush_pending(); + + world_clock &_clock; + world_clock_binding *_binding = nullptr; + + std::vector _step_obs; + std::vector _frame_obs; + std::vector _pending; // observers added during a fan-out + bool _step_dirty = false; + bool _frame_dirty = false; + + std::uint64_t _next_id = 1; + std::uint64_t _next_seq = 0; + std::uint64_t _index = 0; + + bool _started = false; + bool _in_advance = false; + std::chrono::steady_clock::time_point _last_tp; + + clock_frame _frame; +}; + +} // namespace cvc + +#endif // __CVC_CLOCK_DRIVER_H__ diff --git a/src/cvc/CMakeLists.txt b/src/cvc/CMakeLists.txt index c1126f580..bf23c1a10 100644 --- a/src/cvc/CMakeLists.txt +++ b/src/cvc/CMakeLists.txt @@ -54,6 +54,8 @@ set(INCLUDE_FILES ../../inc/cvc/state/state_cluster_membership.h ../../inc/cvc/state/state_hybrid_time.h ../../inc/cvc/core/world_clock.h + ../../inc/cvc/core/clock_follower.h + ../../inc/cvc/core/clock_driver.h ../../inc/cvc/core/world_units.h ../../inc/cvc/core/unit_registry.h ../../inc/cvc/physics/types.h @@ -253,6 +255,7 @@ set(SOURCE_FILES state/state_exec/exec_coordinator.cpp core/world_clock.cpp core/world_clock_binding.cpp + core/clock_driver.cpp core/world_units.cpp core/unit_registry.cpp physics/world.cpp diff --git a/src/cvc/core/clock_driver.cpp b/src/cvc/core/clock_driver.cpp new file mode 100644 index 000000000..cd8edf0a8 --- /dev/null +++ b/src/cvc/core/clock_driver.cpp @@ -0,0 +1,160 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. +*/ + +#include +#include +#include +#include + +namespace cvc { + +clock_driver::clock_driver(world_clock &clock) : _clock(clock) {} + +clock_driver::handle clock_driver::add(std::vector &list, bool &dirty, int order, + observer_fn fn) { + const std::uint64_t id = _next_id++; + entry e; + e.id = id; + e.order = order; + e.seq = _next_seq++; + e.fn = std::move(fn); + e.want_step = (&list == &_step_obs); + if (_in_advance) { + // Added from inside a fan-out: hold it until the current advance finishes so + // it neither fires this frame nor perturbs the list being iterated. + _pending.push_back(std::move(e)); + } else { + list.push_back(std::move(e)); + dirty = true; + } + return handle{id}; +} + +clock_driver::handle clock_driver::add_step_observer(int order, observer_fn fn) { + return add(_step_obs, _step_dirty, order, std::move(fn)); +} + +clock_driver::handle clock_driver::add_frame_observer(int order, observer_fn fn) { + return add(_frame_obs, _frame_dirty, order, std::move(fn)); +} + +bool clock_driver::remove(handle h) { + if (!h) + return false; + for (auto *list : {&_step_obs, &_frame_obs, &_pending}) { + for (auto &e : *list) { + if (e.id == h.id && !e.dead) { + if (_in_advance) { + // Mark it; fire() skips dead entries and the post-advance compaction + // erases them. This lets an observer remove itself or a sibling. + e.dead = true; + } else { + e.dead = true; // compacted below (cheap; keeps one code path) + list->erase( + std::remove_if(list->begin(), list->end(), [](const entry &x) { return x.dead; }), + list->end()); + } + return true; + } + } + } + return false; +} + +void clock_driver::sort_if_dirty(std::vector &list, bool &dirty) { + if (!dirty) + return; + std::stable_sort(list.begin(), list.end(), [](const entry &a, const entry &b) { + if (a.order != b.order) + return a.order < b.order; + return a.seq < b.seq; + }); + dirty = false; +} + +void clock_driver::fire(std::vector &list) { + // Index, not iterators: remove() only marks `dead` while firing and adds go to + // _pending, so `list` is not resized here and the walk stays valid. + for (std::size_t i = 0; i < list.size(); ++i) { + if (!list[i].dead && list[i].fn) + list[i].fn(_frame); + } +} + +void clock_driver::flush_pending() { + auto drop_dead = [](std::vector &l) { + l.erase(std::remove_if(l.begin(), l.end(), [](const entry &x) { return x.dead; }), l.end()); + }; + drop_dead(_step_obs); + drop_dead(_frame_obs); + if (!_pending.empty()) { + for (auto &e : _pending) { + if (e.dead) + continue; + if (e.want_step) { + _step_obs.push_back(std::move(e)); + _step_dirty = true; + } else { + _frame_obs.push_back(std::move(e)); + _frame_dirty = true; + } + } + _pending.clear(); + } +} + +const clock_driver::clock_frame &clock_driver::advance(double wall_dt) { + if (_in_advance) + throw cvc::unsupported_exception("clock_driver::advance re-entered from an observer"); + + double dt; + if (wall_dt < 0.0) { + const auto now = std::chrono::steady_clock::now(); + dt = _started ? std::chrono::duration(now - _last_tp).count() : 0.0; + _last_tp = now; + } else { + dt = wall_dt; + _last_tp = std::chrono::steady_clock::now(); + } + const bool first = !_started; + _started = true; + + const world_clock::step_result r = _binding ? _binding->drive(dt) : _clock.advance(dt); + + _frame.step = r; + _frame.tick = _clock.tick(); + _frame.t = _clock.t(); + _frame.alpha = r.alpha; + _frame.fixed_dt = _clock.fixed_dt(); + _frame.wall_dt = dt; + _frame.first = first; + _frame.index = _index; + + auto finish = [this]() { + _in_advance = false; + flush_pending(); + ++_index; + }; + + _in_advance = true; + try { + sort_if_dirty(_step_obs, _step_dirty); + fire(_step_obs); + sort_if_dirty(_frame_obs, _frame_dirty); + fire(_frame_obs); + } catch (...) { + finish(); + throw; + } + finish(); + return _frame; +} + +} // namespace cvc diff --git a/src/cvc/tests/CMakeLists.txt b/src/cvc/tests/CMakeLists.txt index d64b0396b..3d563ae9a 100644 --- a/src/cvc/tests/CMakeLists.txt +++ b/src/cvc/tests/CMakeLists.txt @@ -104,6 +104,7 @@ add_executable(state_large_tree_bench_test state_large_tree_bench_test.cpp) add_executable(state_hash_partition_test state_hash_partition_test.cpp) add_executable(state_hybrid_time_test state_hybrid_time_test.cpp) add_executable(world_clock_test world_clock_test.cpp) +add_executable(clock_driver_test clock_driver_test.cpp) add_executable(world_units_test world_units_test.cpp) add_executable(physics_world_test physics_world_test.cpp) add_executable(physics_field_test physics_field_test.cpp) @@ -299,6 +300,7 @@ set(TEST_TARGETS state_volume_codec_test state_brick_manifest_test world_clock_test + clock_driver_test world_units_test physics_world_test physics_field_test @@ -1617,6 +1619,13 @@ target_link_libraries(world_clock_test GTest::gtest_main ) +target_link_libraries(clock_driver_test + PRIVATE + cvc + GTest::gtest + GTest::gtest_main +) + target_link_libraries(world_units_test PRIVATE cvc @@ -2026,6 +2035,7 @@ target_compile_features(state_large_tree_bench_test PRIVATE cxx_std_17) target_compile_features(state_hash_partition_test PRIVATE cxx_std_17) target_compile_features(state_hybrid_time_test PRIVATE cxx_std_17) target_compile_features(world_clock_test PRIVATE cxx_std_17) +target_compile_features(clock_driver_test PRIVATE cxx_std_17) target_compile_features(world_units_test PRIVATE cxx_std_17) target_compile_features(physics_world_test PRIVATE cxx_std_17) target_compile_features(physics_field_test PRIVATE cxx_std_17) @@ -2480,6 +2490,7 @@ cvc_discover_tests(state_large_tree_bench_test) cvc_discover_tests(state_hash_partition_test) cvc_discover_tests(state_hybrid_time_test) cvc_discover_tests(world_clock_test) +cvc_discover_tests(clock_driver_test) cvc_discover_tests(world_units_test) cvc_discover_tests(physics_world_test) cvc_discover_tests(physics_field_test) diff --git a/src/cvc/tests/clock_driver_test.cpp b/src/cvc/tests/clock_driver_test.cpp new file mode 100644 index 000000000..c7b41b667 --- /dev/null +++ b/src/cvc/tests/clock_driver_test.cpp @@ -0,0 +1,262 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. +*/ + +// Tests for cvc::clock_driver -- the one-advance-per-frame fan-out over a +// world_clock. The properties worth pinning are the ones a misordered or +// double-fired observer breaks silently: step observers all run before frame +// observers; the clock_frame every observer sees is the SAME reading of time; +// the clock advances exactly once per frame (never per observer); a measured +// first frame books no startup burst; and removing/adding an observer from +// inside the fan-out neither skips a live observer nor fires a new one early. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using cvc::clock_driver; +using cvc::world_clock; + +namespace { +constexpr double k120 = 1.0 / 120.0; +} // namespace + +// ── ordering: all step observers, then all frame observers ────────────────── + +TEST(ClockDriverTest, FiresStepObserversThenFrameObserversEachInAscendingOrder) { + world_clock c; + clock_driver d(c); + std::vector log; + + // Register out of order and across kinds; a frame observer with a lower + // `order` than a step observer must still fire AFTER it (kind wins). + d.add_frame_observer(10, [&](const clock_driver::clock_frame &) { log.push_back("frame10"); }); + d.add_step_observer(5, [&](const clock_driver::clock_frame &) { log.push_back("step5"); }); + d.add_frame_observer(1, [&](const clock_driver::clock_frame &) { log.push_back("frame1"); }); + d.add_step_observer(2, [&](const clock_driver::clock_frame &) { log.push_back("step2"); }); + + d.advance(k120); + + ASSERT_EQ(log.size(), 4u); + EXPECT_EQ(log[0], "step2"); + EXPECT_EQ(log[1], "step5"); + EXPECT_EQ(log[2], "frame1"); + EXPECT_EQ(log[3], "frame10"); +} + +TEST(ClockDriverTest, TiesInOrderKeepRegistrationOrder) { + world_clock c; + clock_driver d(c); + std::vector log; + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { log.push_back("a"); }); + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { log.push_back("b"); }); + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { log.push_back("c"); }); + d.advance(k120); + ASSERT_EQ(log.size(), 3u); + EXPECT_EQ(log[0], "a"); + EXPECT_EQ(log[1], "b"); + EXPECT_EQ(log[2], "c"); +} + +// ── one shared frame reading ──────────────────────────────────────────────── + +TEST(ClockDriverTest, EveryObserverSeesTheSameFrameAndTheClockAdvancesOnce) { + world_clock c; + clock_driver d(c); + double t_a = -1.0, t_b = -1.0; + std::uint64_t tick_a = 0, tick_b = 0; + int calls_a = 0, calls_b = 0; + + d.add_frame_observer(0, [&](const clock_driver::clock_frame &f) { + t_a = f.t; + tick_a = f.tick; + ++calls_a; + }); + d.add_frame_observer(1, [&](const clock_driver::clock_frame &f) { + t_b = f.t; + tick_b = f.tick; + ++calls_b; + }); + + const auto &f = d.advance(3.0 * k120); + + // Advanced exactly once: three whole quanta, not three-per-observer. + EXPECT_EQ(f.step.steps, 3); + EXPECT_EQ(f.tick, 3u); + EXPECT_EQ(c.tick(), 3u); + EXPECT_DOUBLE_EQ(f.t, 3.0 * k120); + // Both observers fired once and read the identical time. + EXPECT_EQ(calls_a, 1); + EXPECT_EQ(calls_b, 1); + EXPECT_DOUBLE_EQ(t_a, t_b); + EXPECT_DOUBLE_EQ(t_a, f.t); + EXPECT_EQ(tick_a, tick_b); +} + +TEST(ClockDriverTest, FrameCarriesStepResultAndAlphaAndWallDt) { + world_clock c; + clock_driver d(c); + const auto &f = d.advance(1.5 * k120); // one whole quantum, half-banked + EXPECT_EQ(f.step.steps, 1); + EXPECT_EQ(f.step.first_tick, 0u); + EXPECT_NEAR(f.alpha, 0.5, 1e-9); + EXPECT_DOUBLE_EQ(f.alpha, f.step.alpha); + EXPECT_DOUBLE_EQ(f.fixed_dt, k120); + EXPECT_DOUBLE_EQ(f.wall_dt, 1.5 * k120); +} + +// ── first-frame seeding ───────────────────────────────────────────────────── + +TEST(ClockDriverTest, MeasuredFirstAdvanceBooksNoTimeThenInjectedAdvancesNormally) { + world_clock c; + clock_driver d(c); + int frames = 0; + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { ++frames; }); + + const auto &f0 = d.advance(); // wall_dt < 0 -> measure; first call seeds only + EXPECT_TRUE(f0.first); + EXPECT_EQ(f0.step.steps, 0); + EXPECT_EQ(f0.tick, 0u); + EXPECT_EQ(f0.index, 0u); + EXPECT_EQ(frames, 1); // observers still fire on the seeding frame + + const auto &f1 = d.advance(2.0 * k120); // injected: advances normally + EXPECT_FALSE(f1.first); + EXPECT_EQ(f1.step.steps, 2); + EXPECT_EQ(f1.tick, 2u); + EXPECT_EQ(f1.index, 1u); +} + +TEST(ClockDriverTest, InjectedDtIsHonoredOnTheFirstCallToo) { + world_clock c; + clock_driver d(c); + const auto &f = d.advance(3.0 * k120); + EXPECT_TRUE(f.first); + EXPECT_EQ(f.step.steps, 3); // injected path does not suppress the first delta +} + +// ── remove / add from inside the fan-out ──────────────────────────────────── + +TEST(ClockDriverTest, OneShotObserverRemovesItselfAndFiresExactlyOnce) { + world_clock c; + clock_driver d(c); + int shots = 0, others = 0; + clock_driver::handle h; + h = d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { + ++shots; + d.remove(h); // self-removal from inside the fan-out + }); + d.add_frame_observer(1, [&](const clock_driver::clock_frame &) { ++others; }); + + d.advance(k120); + d.advance(k120); + d.advance(k120); + + EXPECT_EQ(shots, 1); // fired once, then gone + EXPECT_EQ(others, 3); // the sibling keeps firing every frame +} + +TEST(ClockDriverTest, RemovingASiblingDuringFireSkipsItThisFrame) { + world_clock c; + clock_driver d(c); + int b_fires = 0; + clock_driver::handle hb; + // A (order 0) removes B (order 1) before B would fire this frame. + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { d.remove(hb); }); + hb = d.add_frame_observer(1, [&](const clock_driver::clock_frame &) { ++b_fires; }); + + d.advance(k120); + EXPECT_EQ(b_fires, 0); // removed before its turn, never fires + d.advance(k120); + EXPECT_EQ(b_fires, 0); // stays gone +} + +TEST(ClockDriverTest, AddingAnObserverDuringFireDefersItToTheNextFrame) { + world_clock c; + clock_driver d(c); + int added_fires = 0; + bool armed = true; + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { + if (armed) { + armed = false; + d.add_frame_observer(-100, // even a lower order must not jump into this frame + [&](const clock_driver::clock_frame &) { ++added_fires; }); + } + }); + + d.advance(k120); + EXPECT_EQ(added_fires, 0); // not fired in the frame it was added + d.advance(k120); + EXPECT_EQ(added_fires, 1); // fires from the next frame on +} + +// ── misuse ────────────────────────────────────────────────────────────────── + +TEST(ClockDriverTest, ReenteringAdvanceFromAnObserverThrows) { + world_clock c; + clock_driver d(c); + d.add_frame_observer(0, [&](const clock_driver::clock_frame &) { d.advance(k120); }); + EXPECT_THROW(d.advance(k120), cvc::exception); + // The driver is left usable after the guarded throw. + d.remove(clock_driver::handle{}); // no-op on a null handle +} + +TEST(ClockDriverTest, RemoveReturnsFalseForAnUnknownOrNullHandle) { + world_clock c; + clock_driver d(c); + EXPECT_FALSE(d.remove(clock_driver::handle{})); + auto h = d.add_frame_observer(0, [](const clock_driver::clock_frame &) {}); + EXPECT_TRUE(d.remove(h)); + EXPECT_FALSE(d.remove(h)); // already gone +} + +// ── step observers advance their own sim via a follower ───────────────────── + +TEST(ClockDriverTest, StepObserverRunsDueQuantaViaFrameStep) { + world_clock c; // 120 Hz quantum + clock_driver d(c); + // A sim stepping every 4 quanta: count the ticks it runs across frames. + std::uint64_t sim_steps = 0; + cvc::clock_follower follower(/*period=*/4); + d.add_step_observer( + 0, [&](const clock_driver::clock_frame &f) { sim_steps += follower.take(f.tick); }); + + d.advance(4.0 * k120); // 4 quanta -> exactly one due step + EXPECT_EQ(sim_steps, 1u); + d.advance(8.0 * k120); // 8 more quanta -> two due steps + EXPECT_EQ(sim_steps, 3u); +} + +// ── binding path: sim.* control steers the driver, telemetry is published ──── + +TEST(ClockDriverTest, BindingRoutesControlAndTelemetryThroughTheDriver) { + cvc::app ctx; + world_clock &wc = ctx.world_clock(); // per-app clock, 120 Hz / live + cvc::world_clock_binding b(ctx, wc); // prefix "sim" + clock_driver d(wc); + d.set_binding(&b); + + d.advance(2.0 * k120); + EXPECT_EQ(wc.tick(), 2u); + // Telemetry was pushed to the tree. + EXPECT_EQ(cvc::state::instance(ctx)("sim.tick").value(), std::string("2")); + + // Pause via the control key: pause wins, the driver books nothing. + cvc::state::instance(ctx)("sim.paused").value(true); + d.advance(5.0 * k120); + EXPECT_EQ(wc.tick(), 2u); +} From 7859ba098d1c4c23b401c23d4799c9259c4ba947 Mon Sep 17 00:00:00 2001 From: Joe Rivera Date: Thu, 22 Oct 2026 00:53:38 -0500 Subject: [PATCH 2/3] refactor(arigl): drive the conference pump through cvc::clock_driver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace ConferenceSession's monolithic eleven-step pump() and its hand-rolled media_clock_now() with a single cvc::clock_driver that owns the one per-frame advance of the per-app world_clock and fans the steps out as ordered observers. setup() now builds a clock_driver over app.world_clock() and calls install_observers(), which registers the eleven per-frame steps as frame observers at explicit ascending orders (kOrderPumpAll .. kOrderActivity) — the exact order the open-coded pump ran them in. pump() is now just driver_->advance(): it measures the wall delta, advances the clock once (the single media timestamp, frame.t, handed to forward_streams so a receiver can A/V-align a frame's camera and mic), and fires the observers. The device enumeration becomes a self-removing one-shot observer instead of a per-frame latched branch. Behavior is preserved: same step order, one world-clock reading per frame shared by every step, the first measured advance books no startup burst (as media_clock_now's last_wall_ < 0 skip did), now_wall_ stays a once-per-frame steady_clock reading for the frame-rate-independent talk-hold, and conf.pump still runs before the document on:tick (the deliberate one-frame lag read_peer_controls relies on). A plain local document has no driver and pump() no-ops, exactly as before. This removes the conference's second, hand-rolled world_clock advance path (the map's "never double-advance" hazard) and puts the host on the same one-advance/ordered-observer spine the holon simulation will use. Exercises both the run-mode input pump and the snapshot-mode direct pump. arigl builds clean; the no-relay path (arigl check) is green. --- src/arigl/main.cpp | 117 +++++++++++++++++++++++++++++++-------------- 1 file changed, 81 insertions(+), 36 deletions(-) diff --git a/src/arigl/main.cpp b/src/arigl/main.cpp index 5326bac30..bd1e50b01 100644 --- a/src/arigl/main.cpp +++ b/src/arigl/main.cpp @@ -68,6 +68,7 @@ #include // stream::open — per-peer decode proxies #include // subscribe / publish_external #include // look up the local camera/mic streams by token +#include // one-advance-per-frame fan-out over the world_clock #include // app.world_clock().t() — the common media timestamp #include // list_audio_capture_devices — mic picker #include // list_cameras — camera picker @@ -355,21 +356,31 @@ struct ConferenceSession { return "media." + origin + "." + kind; } - // The common media timestamp: cvc::world_clock seconds (the canonical time base; frame pts is - // meant to be world time). The conference .ari declares no `clock:` block, so nothing else - // advances the per-app clock — drive it from wall time here, once per pump, and stamp BOTH our - // camera and mic frames with the returned t() so a receiver can align them (their sources' own - // pts are on unrelated clocks). t() is tick*fixed_dt (drift-free, 1/120 s quanta — sub-frame for - // media). - double last_wall_ = -1.0; - double media_clock_now() { - const double wall = - std::chrono::duration(std::chrono::steady_clock::now().time_since_epoch()).count(); - if (last_wall_ >= 0.0) - app_->world_clock().advance(wall - last_wall_); - last_wall_ = wall; - return app_->world_clock().t(); - } + // The per-frame clock driver: advances the per-app cvc::world_clock from wall time ONCE per frame + // (the conference .ari declares no `clock:` block, so nothing else drives this app's clock) and + // fans the conference's per-frame steps out as ordered observers. driver_->frame().t is the ONE + // media timestamp per frame — world_clock seconds, tick*fixed_dt (drift-free, 1/120 s quanta; + // sub-frame for media) — stamped on BOTH our camera and mic frames so a receiver can align them + // (their sources' own pts are on unrelated clocks). Built in setup() once the app is known; null + // for a plain local document (no relay), where pump() no-ops. install_observers() defines the + // step order below. + std::unique_ptr driver_; + cvc::clock_driver::handle dev_enum_h_; // handle of the self-removing device-enum one-shot + + // The order each conference step fires in, each frame (ascending). Gaps leave room to slot a + // future step between these without renumbering. + enum : int { + kOrderPumpAll = 10, // drain/flush the state transport (replicated room tree) — first + kOrderWall = 20, // refresh now_wall_ for the wall-time talk-hold + kOrderControls = 30, // read the user-list window's per-peer local controls + kOrderDevEnum = 40, // lazily enumerate camera/mic devices (one-shot) + kOrderApplyDev = 50, // map the device combos' picks to cam/mic.want_device + kOrderForward = 60, // forward our camera+mic, each stamped with frame.t + kOrderDrain = 70, // route peers' inbound frames into per-peer buffers + kOrderMix = 80, // sum peers' queued audio into the single mixed output + kOrderShow = 90, // advance each peer's video to its audio clock + kOrderActivity = 100, // decay + publish talk activity for the roster — last + }; // Call AFTER the GL bundles and BEFORE load(): register the room verbs, seed conf.room/conf.self, // and dial the relay. Returns false (a plain local document) when no relay is configured. @@ -405,9 +416,12 @@ struct ConferenceSession { app_ = &app; pid_ = nodeId; prefix_ = prefix; - // Drive the world clock from wall time (media_clock_now): live mode so advance() banks wall - // time. + // Drive the world clock from wall time via the clock driver: live mode so advance() banks wall + // time. The driver owns the single advance of this app's clock and fires the conference's + // per-frame steps in order (install_observers); nothing else advances it. app.world_clock().set_mode(cvc::world_clock::mode::live); + driver_ = std::make_unique(app.world_clock()); + install_observers(); // The streams we forward: the camera (latest — freshest frame) and the mic (ring — every chunk // in order, since dropped audio is a glitch). The .ari publishes their tokens in // cam_token/mic_token. @@ -492,30 +506,61 @@ struct ConferenceSession { active = true; return true; } + // One advance per frame. The driver advances the per-app world_clock from wall time (the single + // media stamp, frame().t) and fires the conference's per-frame steps as the ordered observers + // install_observers() registered. Called from the input pump (run mode) and directly per frame + // (snapshot mode); a plain local document has no driver and no-ops. void pump() { - if (!tp_) + if (!tp_ || !driver_) return; - tp_->pump_all(); - const double stamp = - media_clock_now(); // one world-clock reading per pump stamps all we forward - now_wall_ = // wall-clock seconds for the talk-hold (frame-rate indep.) - std::chrono::duration(std::chrono::steady_clock::now().time_since_epoch()).count(); - read_peer_controls(); // pull per-peer volume/mute the user-list window set + driver_->advance(); // measured wall delta; fires kOrderPumpAll .. kOrderActivity in order + } + + // Register the per-frame steps as ordered observers on the driver, once, in setup(). They fire + // in ascending kOrder* each advance(), all BEFORE the document's on:tick (conf.pump runs before + // app_rt_.drain()), preserving the deliberate one-frame lag read_peer_controls relies on. + void install_observers() { + using cf = cvc::clock_driver::clock_frame; + // Drive the state transport first: flush replicated room-tree writes and service queued reads + // so room-join/roster are in place before anything this frame reads them. + driver_->add_frame_observer(kOrderPumpAll, [this](const cf &) { tp_->pump_all(); }); + // Wall-clock seconds for the talk-hold — deliberately wall time, not world time, so the hold is + // frame-rate independent and does not freeze when the sim clock is paused or scaled. + driver_->add_frame_observer(kOrderWall, [this](const cf &) { + now_wall_ = std::chrono::duration(std::chrono::steady_clock::now().time_since_epoch()) + .count(); + }); + // Pull the user-list window's per-peer volume/mute (local only, never sent). + driver_->add_frame_observer(kOrderControls, [this](const cf &) { read_peer_controls(); }); // Enumerate the camera/mic devices for the pickers LAZILY — once the camera stream is open, so // the SDL audio/camera subsystems are already initialized (enumeration is then a cheap refcount // bump) and the Asyncify event loop is running. Doing this at setup instead STALLS startup on - // single-threaded wasm (SDL device queries before the loop runs block / can't resolve). - if (!devices_enumerated_ && app_ && - !cvc::state::instance(*app_)(prefix_ + ".cam_token").value().empty()) { - enumerate_devices(); - devices_enumerated_ = true; - } - apply_device_selection(); // map the device combos' picks to cam/mic.want_device - forward_streams(stamp); // (also measures our own mic level for self "talking") - drain_inbox(); // (also measures each peer's mic level) - mix_audio(); // sum all peers' queued audio (per-peer gain, skip muted) - sync_and_show(); // advance each peer's video to its audio clock (skip locally video-muted) - publish_activity(); // decay talk levels + publish media..talking for the name highlight + // single-threaded wasm (SDL device queries before the loop runs block / can't resolve). A + // self-removing one-shot: it checks every frame until the camera token appears, enumerates + // once, then unregisters so the per-frame check stops. + dev_enum_h_ = driver_->add_frame_observer(kOrderDevEnum, [this](const cf &) { + if (!devices_enumerated_ && app_ && + !cvc::state::instance(*app_)(prefix_ + ".cam_token").value().empty()) { + enumerate_devices(); + devices_enumerated_ = true; + driver_->remove(dev_enum_h_); + } + }); + // Map the device combos' picks to cam/mic.want_device for the .ari to re-capture against. + driver_->add_frame_observer(kOrderApplyDev, [this](const cf &) { apply_device_selection(); }); + // Forward our camera (latest) + mic (ring), each re-stamped with THIS frame's single + // world-clock reading (frame.t) so a receiver can A/V-align them; also measures our own mic + // level. + driver_->add_frame_observer(kOrderForward, [this](const cf &f) { forward_streams(f.t); }); + // Route peers' inbound frames (queued by the reader thread) into per-peer buffers; also + // measures each peer's mic level. + driver_->add_frame_observer(kOrderDrain, [this](const cf &) { drain_inbox(); }); + // Sum all peers' queued audio into the single mixed output (per-peer gain, skip muted). + driver_->add_frame_observer(kOrderMix, [this](const cf &) { mix_audio(); }); + // Advance each peer's video to its audio clock (skip locally video-muted). + driver_->add_frame_observer(kOrderShow, [this](const cf &) { sync_and_show(); }); + // Decay talk levels + publish media..talking for the name highlight. + driver_->add_frame_observer(kOrderActivity, [this](const cf &) { publish_activity(); }); } // Pull the per-peer LOCAL controls the user-list window writes — keyed by roster INDEX, which the From 535f39734a8c115a4365bb97c8df6fe14d3db75d Mon Sep 17 00:00:00 2001 From: Joe Rivera Date: Thu, 22 Oct 2026 01:06:32 -0500 Subject: [PATCH 3/3] fix(arigl): join the conference transport reader before teardown (RAII) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConferenceSession had no destructor, so its members were destroyed in reverse declaration order: media_mu_ and inbox_ (declared after the transport) were torn down BEFORE ws_/grpc_, whose destructor is the only thing that joins the transport reader thread. On every teardown path that does not reach the run loop's explicit conf.stop() — snapshot, check, info, a document-load failure, or an exception unwinding the run loop — the still-running reader could invoke on_media and lock a destroyed media_mu_ / push to a destroyed inbox_: a use-after-free and data race. The snapshot-of-a-live-conference path is explicitly supported, so this was reachable, not theoretical. Add ~ConferenceSession() { stop(); }. The destructor body runs ahead of member destruction, so stop() joins the reader (via stop_(), which the transport makes idempotent with _running.exchange(false)) while media_mu_/inbox_ are still alive. ConferenceSession::stop() is already idempotent (clear/reset are no-ops the second time), so the run path's explicit stop() followed by this one is safe. Single-threaded wasm has no reader thread and is unaffected. Pre-existing hazard (not introduced by the clock_driver refactor); surfaced by an adversarial review of this branch and fixed here since it lives in the same struct the refactor touches. --- src/arigl/main.cpp | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/arigl/main.cpp b/src/arigl/main.cpp index bd1e50b01..c7eef3de1 100644 --- a/src/arigl/main.cpp +++ b/src/arigl/main.cpp @@ -669,9 +669,19 @@ struct ConferenceSession { for (publisher &p : pubs_) p.sub.reset(); if (stop_) - stop_(); + stop_(); // joins the transport reader thread (idempotent: _running.exchange(false)) } + // Join the transport reader BEFORE any member is destroyed. stop() runs in the destructor body, + // ahead of reverse-order member destruction, so the reader (which the transport destructor would + // otherwise join only AFTER media_mu_/inbox_ are already gone — they are declared after + // ws_/grpc_) can no longer fire on_media into a destroyed mutex/vector. This covers every + // teardown path: snapshot/check/info and a document-load failure return without the run loop's + // explicit stop(), and an exception unwinding the run loop skips it too. stop() is idempotent, so + // the run path's explicit call followed by this one is safe. (Single-threaded wasm has no reader + // thread.) + ~ConferenceSession() { stop(); } + // Mix: pop one chunk from each peer that has audio queued, sum them (silence for an empty peer), // clamp, and play to the single output. Produce mixed chunks while the output is below ~100 ms so // it stays fed without running far ahead; stop when no peer has a chunk (output drains to