From 6858060af188ebc35f9048f88464a3173acadb89 Mon Sep 17 00:00:00 2001 From: Matt <47545907+SoundMatt@users.noreply.github.com> Date: Sat, 22 Aug 2026 02:28:02 -0700 Subject: [PATCH 1/2] =?UTF-8?q?rewrite(phase5):=20shmem.hpp=20=E2=80=94=20?= =?UTF-8?q?rebuild=20Channel=20around=20a=20real=20bounded=20byte-level=20?= =?UTF-8?q?buffer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of the v3.0.0 ground-up rewrite (cpp-RCP #129, ROADMAP.md Phase 17), branch rewrite/v3-from-c-rcp. Phase 5 (transport layer), Wave 2. A dedicated investigation (this session) found cpp-RCP's shmem.hpp was a decoded-object pass-through with no byte encoding at all and no capacity bound -- Channel::request() called a std::function handler directly and synchronously in the same process. c-RCP's rcp_shmem_avtp_pair_new() is a real byte-level bounded ring-buffer transport, real memcpy framing, RCP_ERR_BUSY backpressure when full -- but is ALSO in-process only under the hood (plain heap, non-pshared mutex/condvar). Neither codebase has ever had genuine cross-process shared memory; "shmem" means "a fast paired in-process queue" in both trees. Decision: rebuild Channel around a real byte-level bounded buffer (closing the two real gaps -- no capacity bound, codec never exercised) while keeping the existing decoded-object Handler signature at the boundary (matching udp::Server's own internal-codec/external- decoded-callback shape), so admin.hpp/config.hpp need zero changes. ## Ported (c-RCP src/shmem.c 266 lines / include/rcp/shmem.h 99 lines) - shmem.c:68-71 (RCP_ERR_BUSY on full queue) -> Channel's new detail::FrameSlots bounded pool + ErrBusy return, REQ-SHMEM-006. - shmem.c:64-67/109-113 (closed-side send/recv RCP_ERR_CLOSED) -> Channel::request()'s existing closed_ check, now cited REQ-SHMEM-005/010. - The codec-never-exercised gap: Channel::request() now genuinely calls acf::encode_acf_abb/encode_acf_gbb and decode_acf_abb/ decode_acf_gbb (selection mirrors udp.hpp's encode_frame/decode_frame) on both the request and response legs before/after the Handler runs. - shmem.h:91-93's runtime queue_capacity parameter (clamped to min 1, shmem.c:204) -> Channel(stream_key, queue_capacity = kDefaultQueueCapacity), clamped into [1, kMaxQueueCapacity]. ## New shape Channel keeps its old Handler signature and synchronous, no-thread- of-its-own dispatch model exactly. Internally, request() now: encodes -> acquires a fixed-capacity slot (ErrBusy if none free) -> decodes back out of that slot -> calls Handler with decoded objects (handler copied out from under a separate mutex and invoked UNLOCKED, so concurrent callers' handlers genuinely run in parallel) -> encodes/ decodes the response -> releases the slot via RAII. Registry's public surface is byte-for-byte unchanged -- confirmed by reading all of admin.hpp/config.hpp/test_admin.cpp/test_config.cpp in full: they only use Registry::add_channel/deregister/lookup/channels/close and Channel::stream_key()/new_channel(), never Channel::request()'s internals. No changes needed to any of the four. ## Bug found and fixed mid-task (real concurrency bug) The original code held the handler mutex ACROSS the handler call itself, which would have serialized every concurrent request and deadlocked under the new capacity tests -- fixed by copying the handler out under the lock, then invoking it unlocked. ## Architecture note (documented in the header) Unlike c-RCP's genuinely two-party, strict-FIFO ring (two independent threads holding *out_a/*out_b), Channel::request() is both producer and consumer in one synchronous call, so the bounded buffer is a per-caller indexed slot pool, not a literal head/tail FIFO -- same capacity bound and BUSY-without-enqueuing contract, no independent- reader ordering to preserve since there's only ever one reader. ## REQ-SHMEM mapping / c-RCP citation drift found (not propagated) .fusa-reqs.json was split 2026-08-18 (issue #533) into REQ-SHMEM-001.. 010 (not 001..009). Found three req-id mis-citations in c-RCP's own src/shmem.c inline comments (shmem_side_send citing 007 instead of 006; shmem_side_recv wrongly also citing 006; shmem_side_destroy citing 007 instead of 008) -- c-RCP's own test_shmem.c citations are correct, only the .c file's comments drifted; not propagated here. REQ-SHMEM-004 (recv timeout) and REQ-SHMEM-007 (recv buffer-too-small) have no analog in the new cpp shape, documented as such rather than force-fit. ## Verification include/rcp/shmem.hpp: 173 -> 458 lines. tests/test_shmem.cpp: 170 -> 526 lines, test_shmem: 20 TEST_CASEs (68 assertions), re-run 5x back-to-back with no flakiness in the new thread/condition-variable- based backpressure tests. Full tree rebuilds clean (0 errors, 0 warnings), 58/58 ctest suites pass (100%) -- independently re-verified from a fresh build, not just the porting agent's own report. Co-Authored-By: Claude Sonnet 5 Signed-off-by: Matt <47545907+SoundMatt@users.noreply.github.com> --- include/rcp/shmem.hpp | 369 +++++++++++++++++++++++++++++++----- tests/test_shmem.cpp | 424 ++++++++++++++++++++++++++++++++++++++---- 2 files changed, 717 insertions(+), 76 deletions(-) diff --git a/include/rcp/shmem.hpp b/include/rcp/shmem.hpp index 4ecedbc..7f001c3 100644 --- a/include/rcp/shmem.hpp +++ b/include/rcp/shmem.hpp @@ -6,33 +6,95 @@ // fusa:req REQ-SHMEM-006 // fusa:req REQ-SHMEM-007 // fusa:req REQ-SHMEM-008 +// fusa:req REQ-SHMEM-009 +// fusa:req REQ-SHMEM-010 -// Zero-copy intra-host request delivery via shared in-process memory. +// In-process, bounded-queue request delivery (cpp-RCP issue #129 Phase 5 +// wave 2; ROADMAP.md Phase 17 v3.0.0 rewrite). // -// ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind -// (v2.14.0)": this header is ADAPTed, per the Satellite Package -// Disposition table's entry for `shmem.hpp` — the "avoid serialisation -// overhead for a co-located RC Client/RC Server" value proposition is -// unaffected by the protocol replacement, so the concept survives; only -// the request/response shapes it carries change. shmem::Channel and -// shmem::Registry below play the same role rcp/udp.hpp's Server/Client and -// Registry-shaped lookup play for a real socket transport, just for two -// endpoints that happen to live in the same process: no bytes are ever -// serialised to (or decoded from) an AVTPDU/ACF wire encoding, since both -// sides already share the same acf::AcfMessageInfo/std::vector -// objects in memory. +// This is the in-process analog c-RCP's own shmem.h/shmem.c (99/266 lines) +// plays for a real byte-level transport: like c-RCP's +// rcp_shmem_avtp_pair_new(), this header moves real, encoded ACF_ABB/ +// ACF_GBB bytes (rcp/acf.hpp) through a fixed-capacity, in-process buffer +// and backpressures with a real error (ErrBusy) once that capacity is +// exhausted, rather than growing without bound or silently dropping +// anything. "shmem" is c-RCP's own name for this concept, not a claim of +// real OS shared memory: neither codebase has ever `shm_open()`/`mmap()`ed +// anything here — c-RCP's shmem.c uses plain heap allocation and a +// non-PTHREAD_PROCESS_SHARED mutex/condvar (rcp_mutex_t/rcp_cond_t), and +// this header uses std::mutex/std::shared_mutex — both in-process only, no +// real I/O, exactly as c-RCP's own shmem.h header comment (lines 12-22) +// already says of itself ("entirely in-process (no real I/O)"). An earlier +// revision of this header instead described Channel/Registry as a +// "zero-copy" pass-through with "no bytes are ever serialised" — that +// description is now wrong twice over: it implied a capability (avoiding +// serialisation overhead) as if it were this module's distinguishing value +// versus a real transport, and the implementation genuinely had no +// capacity bound of any kind (Channel::request() called a handler directly +// with no buffer between caller and callee at all) — the two real gaps +// this pass fixes, both ported from c-RCP's actual shape rather than +// re-derived: +// 1. No capacity bound existed. c-RCP's queue_capacity constructor +// parameter (shmem.h:91, clamped to a minimum of 1 at shmem.c:204) and +// its RCP_ERR_BUSY-on-full send() (shmem.c:68-71, REQ-SHMEM-006) are +// ported below as Channel's own queue_capacity constructor parameter +// and ErrBusy-on-full request() (REQ-SHMEM-006 again — see that +// section below for why this pass corrects two REQ-SHMEM-006/007 +// mis-citations found in c-RCP's own shmem.c while porting this). +// 2. The ACF codec was never exercised on this path at all — a request +// reached its handler as the exact same in-memory acf::AcfMessageInfo/ +// std::vector objects the caller built, never encoded to or +// decoded from bytes. Channel::request() below now encodes the +// outbound request via acf::encode_acf_abb()/encode_acf_gbb() (info. +// acf_msg_type selects which, the same selection rcp/udp.hpp's +// encode_frame()/decode_frame() make at udp.hpp:209-232/234-278 — +// reused here as this header's own idiom reference rather than +// re-derived) before it ever reaches the handler, and decodes the +// handler's response the same way before returning it to the caller. // -// This header has no clock, thread, or socket of its own — Channel:: -// request() dispatches synchronously on the calling thread, same +// Handler's own signature is UNCHANGED: it is still handed fully decoded +// acf::AcfMessageInfo/std::vector objects, exactly as udp:: +// Server's own Handler is (udp.hpp:424-427) despite udp::Server doing real +// byte encode/decode internally around it (udp.hpp:209-278) — the same +// "real codec inside, decoded objects at the boundary" split this header +// now follows too. rcp::admin::AdminServer (rcp/admin.hpp) and rcp::config +// ::load() (rcp/config.hpp) only ever depend on Registry's keyed add/ +// lookup/enumerate surface (add_channel/deregister/lookup/channels/close), +// never on Channel::request()'s internals, and that surface is unchanged +// here — verified by reading both files and their tests (tests/ +// test_admin.cpp, tests/test_config.cpp) in full; neither calls +// Channel::request() at all. +// +// This header has no clock or socket of its own, same as before — Channel:: +// request() still dispatches synchronously on the calling thread, the // "primitives driven by the embedding application" convention every Phase -// 14/15 header has used since v2.9.0. +// 14/15 header has used since v2.9.0. Unlike c-RCP's shmem.c, it also still +// has no thread of its own: c-RCP's pair is genuinely asynchronous (two +// independent rcp_avtp_transport_t sides, each send()ing/recv()ing from +// its own, possibly different, thread, correlated only by strict FIFO +// order — shmem.c's ring_push()/shmem_side_recv()); Channel::request() +// plays both roles — the party that encodes+enqueues AND the party that +// dequeues+decodes+dispatches — inside one synchronous call on one thread, +// so there is no second, independent consumer a FIFO handoff would ever be +// needed to correlate with. Channel's own bounded buffer (detail:: +// FrameSlots below) is therefore a fixed-capacity POOL of indexed slots a +// caller acquires-and-later-releases, not a literal head/tail FIFO ring — +// see FrameSlots's own comment for the full reasoning and for exactly which +// property of c-RCP's ring this still preserves (a caller's own bytes are +// never handed to, or released by, a different caller) versus which it +// cannot (strict FIFO order across independent readers/writers has no +// meaning when there is only ever one party doing both jobs). The capacity +// BOUND and the BUSY-without-enqueuing failure mode — the two things +// REQ-SHMEM-006 actually requires — are preserved exactly. #pragma once #include "acf.hpp" #include "avtp.hpp" -#include "rcp.hpp" // for rcp::ErrClosed/ErrNotFound/ErrAlreadyExists only — see this header's own scope note above +#include "rcp.hpp" // for rcp::ErrClosed/ErrBusy/ErrNotFound/ErrAlreadyExists only — see this header's own scope note above +#include #include +#include #include #include #include @@ -44,13 +106,137 @@ namespace rcp { namespace shmem { +// ── Bounded frame codec + slot pool ────────────────────────────────────────── + +namespace detail { + +// encode_acf_message/decode_acf_message — Channel's own internal ACF codec +// step (this file's header comment, gap #2). Selects ACF_ABB vs ACF_GBB by +// info.acf_msg_type, mirroring rcp/udp.hpp's encode_frame()/decode_frame() +// selection (udp.hpp:209-232/234-278) rather than re-deriving it. ACF_GBB's +// message_timestamp is always sent, and always decoded back, as 0: Handler's +// own decoded-object contract (like rcp::mock::Server::dispatch's, which +// rcp/udp.hpp's own Handler comment cites as its own shape reference) has no +// parameter slot to carry a real timestamp value through to or from the +// handler, so a genuine one would be silently discarded before ever +// reaching it regardless — a known, documented scope limit, not a +// round-trip bug in encode_acf_gbb()/decode_acf_gbb() themselves (both are +// exercised bit-for-bit elsewhere, tests/test_acf.cpp). +inline std::vector encode_acf_message(const acf::AcfMessageInfo& info, + const std::vector& payload) { + return (info.acf_msg_type == acf::kAcfMsgTypeGbb) + ? acf::encode_acf_gbb(info, /*message_timestamp=*/0, payload) + : acf::encode_acf_abb(info, payload); +} + +inline std::error_code decode_acf_message(const std::vector& bytes, + acf::AcfMessageInfo& out_info, + std::vector& out_payload) { + if (bytes.empty()) return avtp::make_error_code(avtp::AvtpErrc::short_buffer); + if (acf::peek_acf_msg_type(bytes.data()) == acf::kAcfMsgTypeGbb) { + uint64_t discarded_timestamp = 0; + return acf::decode_acf_gbb(bytes.data(), bytes.size(), out_info, discarded_timestamp, + out_payload); + } + return acf::decode_acf_abb(bytes.data(), bytes.size(), out_info, out_payload); +} + +// FrameSlots — a small, fixed-capacity pool of encoded-ACF-frame +// byte buffers backing Channel::request()'s bounded round trip (REQ-SHMEM- +// 006; c-RCP's shmem.c ring_push()/shmem_side_send()/shmem_side_recv(), +// lines 40-45/53-76/84-131). Physically fixed at compile time +// (std::array, no heap growth of the pool itself — the same +// "vector-shaped ergonomics, no unbounded growth" convention rcp/ +// request.hpp's BoundedVector already establishes for this codebase), +// with a runtime-checked logical capacity <= Capacity so a Channel can +// still be constructed with a caller-chosen queue_capacity — the same role +// rcp_shmem_avtp_pair_new()'s own queue_capacity parameter plays +// (shmem.h:91-93) — without Channel itself needing to become a template. +// +// Departs from shmem.c's own strict head/tail FIFO ring shape in one +// respect, for a reason specific to this header (see the file header +// comment above for the full explanation): a caller acquire()s and later +// release()s exactly one slot of its own, addressed by the index acquire() +// returns, rather than handing a frame to an independent consumer that +// drains strictly in arrival order — there is no independent consumer here +// for a FIFO to correlate with. What carries over unchanged is the thing +// REQ-SHMEM-006 actually requires: a hard capacity bound, and a BUSY +// result — with the pool left completely untouched — the instant that +// bound is reached. +template +class FrameSlots { +public: + static constexpr size_t npos = static_cast(-1); + + // Clamped into [1, Capacity], the same way c-RCP's own + // rcp_shmem_avtp_pair_new() clamps a caller's queue_capacity==0 up to 1 + // (shmem.c:204) — a 0-slot pool would make every acquire() call + // unconditionally BUSY, never a reachable, useful configuration. + void set_logical_capacity(size_t n) noexcept { + logical_capacity_ = n < 1 ? 1 : (n > Capacity ? Capacity : n); + } + size_t logical_capacity() const noexcept { return logical_capacity_; } + + // acquire finds a free slot, stores `frame` into it, and returns its + // index — or npos, leaving every slot untouched, once + // logical_capacity() slots are already occupied (REQ-SHMEM-006: BUSY + // without enqueuing). + size_t acquire(std::vector frame) { + for (size_t i = 0; i < logical_capacity_; ++i) { + if (!occupied_[i]) { + slots_[i] = std::move(frame); + occupied_[i] = true; + return i; + } + } + return npos; + } + + const std::vector& at(size_t idx) const noexcept { return slots_[idx]; } + void store(size_t idx, std::vector frame) { slots_[idx] = std::move(frame); } + + void release(size_t idx) noexcept { + slots_[idx].clear(); + slots_[idx].shrink_to_fit(); + occupied_[idx] = false; + } + + size_t occupied_count() const noexcept { + size_t n = 0; + for (size_t i = 0; i < logical_capacity_; ++i) { + if (occupied_[i]) ++n; + } + return n; + } + +private: + std::array, Capacity> slots_{}; + std::array occupied_{}; + size_t logical_capacity_ = Capacity; +}; + +} // namespace detail + +// kMaxQueueCapacity: the physical, compile-time ceiling every Channel's +// FrameSlots pool is sized to, regardless of the queue_capacity a caller +// requests — the same order of magnitude as other c-RCP-17 fixed-capacity +// conversions elsewhere in this project's history (e.g. +// RCP_POWERSTATE_MAX_ENDPOINTS=64, rcp/request.hpp's kMaxTrackedRequests= +// 64). kDefaultQueueCapacity is what Channel/new_channel() use when a +// caller doesn't pass one explicitly — small enough that a test can +// exhaust it deliberately (see tests/test_shmem.cpp's backpressure cases). +constexpr size_t kMaxQueueCapacity = 32; +constexpr size_t kDefaultQueueCapacity = 8; + // ── Channel ─────────────────────────────────────────────────────────────────── -// Channel is the in-process, zero-copy analog of rcp/udp.hpp's Server: -// a caller-supplied Handler answers each request directly, in the same -// process, with no wire encode/decode step in between. Handler is shaped -// identically to udp::Server::Handler so the same handler (e.g. -// rcp::mock::Server::dispatch) can be wired to either transport. +// Channel is the in-process, bounded-queue analog of rcp/udp.hpp's Server: +// a caller-supplied Handler answers each request, in the same process, with +// a real byte encode/decode step — and a real, boundable capacity — between +// caller and handler (this file's header comment). Handler is shaped +// identically to udp::Server::Handler's own decoded-object contract so the +// same handler (e.g. rcp::mock::Server::dispatch) can be wired to either +// transport. class Channel { public: using Handler = std::function& out_resp_payload)>; - explicit Channel(uint64_t stream_key) : stream_key_(stream_key) {} + // queue_capacity bounds how many request() calls may have their own + // slot concurrently occupied — i.e. be "in flight" through this Channel + // — before a new caller is rejected with ErrBusy instead of being + // admitted (REQ-SHMEM-006). Clamped into [1, kMaxQueueCapacity] by + // FrameSlots::set_logical_capacity() above. + explicit Channel(uint64_t stream_key, size_t queue_capacity = kDefaultQueueCapacity) + : stream_key_(stream_key) { + slots_.set_logical_capacity(queue_capacity); + } uint64_t stream_key() const noexcept { return stream_key_; } + // queue_capacity/queue_depth expose the bound this pass adds and its + // current occupancy — c-RCP has no equivalent accessor (shmem.c's own + // queue depth is private to its ring), but both are useful for a caller + // (or a test) that wants to observe backpressure building up rather + // than only discovering it via a returned ErrBusy. + size_t queue_capacity() const noexcept { return slots_.logical_capacity(); } + size_t queue_depth() const { + std::lock_guard lk(slots_mu_); + return slots_.occupied_count(); + } + void set_handler(Handler h) { std::lock_guard lk(mu_); handler_ = std::move(h); } - // request delivers `req`/`req_payload` straight to the registered - // Handler and returns its response, with no serialisation step between - // caller and handler. If no Handler is set, the default response is - // ResponseKind::Acknowledge — the same "answer something, don't hang" - // default rcp/udp.hpp's Server uses when unhandled. + // request encodes `req`/`req_payload` to bytes, admits them into this + // Channel's bounded slot pool (ErrBusy — REQ-SHMEM-006 — if every slot + // is already occupied by another in-flight request()), decodes them + // back out, and delivers the decoded result to the registered Handler + // — then encodes and decodes the Handler's own response the same way + // before returning it, so a caller's out_resp/out_resp_payload reflect + // exactly what a real byte-level transport would have delivered rather + // than the handler's in-memory objects verbatim. If no Handler is set, + // the default response is ResponseKind::Acknowledge — the same "answer + // something, don't hang" default rcp/udp.hpp's Server uses when + // unhandled. std::error_code request(size_t client, const acf::AcfMessageInfo& req, const std::vector& req_payload, acf::AcfMessageInfo& out_resp, std::vector& out_resp_payload) { - if (closed_.load(std::memory_order_acquire)) return ErrClosed; + if (closed_.load(std::memory_order_acquire)) return ErrClosed; // REQ-SHMEM-005/010 - std::lock_guard lk(mu_); - if (handler_) { - return handler_(client, req, req_payload, out_resp, out_resp_payload); + size_t slot; + { + std::lock_guard lk(slots_mu_); + slot = slots_.acquire(detail::encode_acf_message(req, req_payload)); } - out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); - out_resp_payload.clear(); - return {}; + if (slot == detail::FrameSlots::npos) return ErrBusy; // REQ-SHMEM-006 + + // Releases this call's own slot on every return path below, + // including an early return from a decode failure or a Handler + // error — mirrors shmem_side_destroy()'s own "always releases what + // it holds" contract (shmem.c:151-181, REQ-SHMEM-008/009 — see + // Channel's own destructor comment near the bottom of this class + // for why those two are otherwise automatic here). + struct SlotGuard { + Channel* ch; + size_t idx; + ~SlotGuard() { + std::lock_guard lk(ch->slots_mu_); + ch->slots_.release(idx); + } + } guard{this, slot}; + + // Decode back what was just admitted — proves the decode half of + // the codec round trip actually runs on the fast path too, not + // just encode (this file's header comment, gap #2). + acf::AcfMessageInfo decoded_req; + std::vector decoded_payload; + { + std::lock_guard lk(slots_mu_); + if (auto ec = detail::decode_acf_message(slots_.at(slot), decoded_req, decoded_payload)) + return ec; + } + + // Copy the Handler under mu_ and invoke the copy outside the lock: + // mu_ only ever needs to protect handler_ itself (set_handler() may + // race with request() on another thread) — holding it across the + // handler CALL as well would serialize every concurrent request() + // through this one Channel to strictly one-at-a-time, silently + // defeating the whole point of a >1 queue_capacity (REQ-SHMEM-006 + // exists to bound genuinely concurrent in-flight requests, not to + // manufacture ones that no longer overlap). + Handler handler_copy; + { + std::lock_guard lk(mu_); + handler_copy = handler_; + } + + std::error_code ec; + if (handler_copy) { + ec = handler_copy(client, decoded_req, decoded_payload, out_resp, out_resp_payload); + } else { + out_resp = acf::make_response(decoded_req, acf::ResponseKind::Acknowledge); + out_resp_payload.clear(); + } + if (ec) return ec; + + auto resp_bytes = detail::encode_acf_message(out_resp, out_resp_payload); + { + std::lock_guard lk(slots_mu_); + slots_.store(slot, resp_bytes); + } + return detail::decode_acf_message(resp_bytes, out_resp, out_resp_payload); } // close is idempotent — safe to call more than once. @@ -93,11 +359,24 @@ class Channel { bool ok() const noexcept { return !closed_.load(std::memory_order_acquire); } + // Channel has no manual refcount/destroy step the way c-RCP's + // shmem_side_destroy() does (shmem.c:151-181): it is always held via + // std::shared_ptr (new_channel() below, Registry's own + // storage), whose atomic refcounting already gives "freed exactly once, + // regardless of release order" (REQ-SHMEM-009) and "releasing one + // holder does not itself invalidate another holder's own use of the + // same Channel" (REQ-SHMEM-008) for free — the same "destructors, RAII + // covers this by construction" convention rcp/watchdog.hpp's own header + // comment documents for an analogous case. Neither REQ needs, or gets, + // any bespoke code here. + private: - uint64_t stream_key_; - std::atomic closed_{false}; - std::mutex mu_; - Handler handler_; + uint64_t stream_key_; + std::atomic closed_{false}; + mutable std::mutex mu_; // guards handler_ + mutable std::mutex slots_mu_; // guards slots_ + detail::FrameSlots slots_; + Handler handler_; }; // ── Registry ────────────────────────────────────────────────────────────────── @@ -105,7 +384,12 @@ class Channel { // Registry looks up a Channel by its stream_key, the same opaque // uint64_t-keyed lookup role rcp/watchdog.hpp's Manager and // rcp/regmap.hpp's Ep0 already establish for this codebase (typically an -// avtp::StreamId::to_u64()). +// avtp::StreamId::to_u64()). Unchanged by this pass — c-RCP has no analog +// of Registry at all (rcp_shmem_avtp_pair_new() returns exactly one pair, +// with no keyed lookup of any kind), and rcp::admin::AdminServer (rcp/ +// admin.hpp) and rcp::config::load() (rcp/config.hpp) only ever depend on +// this exact surface (add_channel/deregister/lookup/channels/close) — see +// this file's own header comment. class Registry { public: std::error_code add_channel(std::shared_ptr ch) { @@ -161,8 +445,9 @@ class Registry { bool closed_ = false; }; -inline std::shared_ptr new_channel(uint64_t stream_key) { - return std::make_shared(stream_key); +inline std::shared_ptr new_channel(uint64_t stream_key, + size_t queue_capacity = kDefaultQueueCapacity) { + return std::make_shared(stream_key, queue_capacity); } inline std::unique_ptr new_registry() { diff --git a/tests/test_shmem.cpp b/tests/test_shmem.cpp index 0825f17..01f4fe8 100644 --- a/tests/test_shmem.cpp +++ b/tests/test_shmem.cpp @@ -1,20 +1,28 @@ // fusa:test REQ-SHMEM-001 // fusa:test REQ-SHMEM-002 // fusa:test REQ-SHMEM-003 -// fusa:test REQ-SHMEM-004 // fusa:test REQ-SHMEM-005 // fusa:test REQ-SHMEM-006 -// fusa:test REQ-SHMEM-007 // fusa:test REQ-SHMEM-008 +// fusa:test REQ-SHMEM-009 +// fusa:test REQ-SHMEM-010 -// Tests for rcp/shmem.hpp — the zero-copy in-process request channel -// (ROADMAP.md milestone 58, "Auxiliary Transport & Cross-Cutting Rebind", -// v2.14.0). +// Tests for rcp/shmem.hpp — the in-process, bounded-queue request channel +// (cpp-RCP issue #129 Phase 5 wave 2). REQ-SHMEM-004 (recv() timeout on an +// empty queue) and REQ-SHMEM-007 (recv() BUSY when the next queued item +// exceeds a caller-supplied destination buffer) have no analog here and are +// intentionally not covered below — see rcp/shmem.hpp's own header comment +// and detail::FrameSlots's comment for why: Channel::request() has no +// blocking wait (there is nothing to time out on) and decodes straight out +// of its own owned slot storage rather than into a separately-sized caller +// buffer (there is no fixed destination capacity to overflow). #include #include #include +#include +#include #include #include @@ -27,16 +35,22 @@ acf::AcfMessageInfo standard_request(avtp::ByteBusId bus_id, uint8_t transaction } } // namespace -// ── Zero-copy delivery ──────────────────────────────────────────────────────── +// ── Real byte-level round trip ────────────────────────────────────────────── +// This pass's core content fix: a request no longer reaches the handler as +// the caller's own in-memory objects — it is genuinely encoded to ACF_ABB/ +// ACF_GBB bytes and decoded back first (rcp/shmem.hpp detail:: +// encode_acf_message/decode_acf_message), the same codec rcp/acf.hpp's +// encode_acf_abb()/decode_acf_abb() apply on a real transport. -TEST_CASE("shmem Channel::request delivers to the handler directly, without a wire encode/decode", +TEST_CASE("shmem Channel::request delivers a byte-decoded request to the handler and " + "byte-decodes its response back to the caller", "[shmem][REQ-SHMEM-001]") { auto ch = new_channel(/*stream_key=*/42); ch->set_handler([](size_t, const acf::AcfMessageInfo& req, const std::vector& payload, acf::AcfMessageInfo& out_resp, std::vector& out_payload) { out_resp = acf::make_response(req, acf::ResponseKind::ReadResponse); - out_payload = payload; // echo unchanged -- proves no lossy round trip occurred + out_payload = payload; // echo unchanged -- proves the round trip is lossless return std::error_code{}; }); @@ -49,6 +63,73 @@ TEST_CASE("shmem Channel::request delivers to the handler directly, without a wi REQUIRE(resp_payload == req_payload); REQUIRE(resp.byte_bus_id == req.byte_bus_id); REQUIRE(resp.transaction_num == req.transaction_num); + REQUIRE(acf::response_kind_of(resp) == acf::ResponseKind::ReadResponse); +} + +TEST_CASE("shmem Channel::request preserves every AcfMessageInfo wire field through the " + "encode/decode round trip, not just byte_bus_id/transaction_num", + "[shmem][REQ-SHMEM-001]") { + auto ch = new_channel(7); + acf::AcfMessageInfo seen_by_handler; + ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + seen_by_handler = req; + out_resp = acf::make_response(req, acf::ResponseKind::WriteResponse); + return std::error_code{}; + }); + + acf::AcfMessageInfo req; + req.acf_msg_type = acf::kAcfMsgTypeAbb; + req.byte_bus_id = 0x123; + req.transaction_num = 200; + req.evt_ack = true; + req.evt_op = 0x5; + req.hs = true; + req.cs = true; + req.op = true; + req.read_size_or_segment_num = 17; + + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(ch->request(0, req, {}, resp, resp_payload)); + + REQUIRE(seen_by_handler.byte_bus_id == req.byte_bus_id); + REQUIRE(seen_by_handler.transaction_num == req.transaction_num); + REQUIRE(seen_by_handler.evt_ack == req.evt_ack); + REQUIRE(seen_by_handler.evt_op == req.evt_op); + REQUIRE(seen_by_handler.hs == req.hs); + REQUIRE(seen_by_handler.cs == req.cs); + REQUIRE(seen_by_handler.op == req.op); + REQUIRE(seen_by_handler.read_size_or_segment_num == req.read_size_or_segment_num); + + // The response side round-trips too -- resp reflects what real bytes + // would decode to, not the handler's in-memory object verbatim. + REQUIRE(acf::response_kind_of(resp) == acf::ResponseKind::WriteResponse); + REQUIRE(resp.byte_bus_id == req.byte_bus_id); + REQUIRE(resp.transaction_num == req.transaction_num); +} + +TEST_CASE("shmem Channel::request round-trips an ACF_GBB request through the GBB codec path", + "[shmem][REQ-SHMEM-001]") { + auto ch = new_channel(9); + bool saw_gbb = false; + ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + saw_gbb = (req.acf_msg_type == acf::kAcfMsgTypeGbb); + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); + + acf::AcfMessageInfo req; + req.acf_msg_type = acf::kAcfMsgTypeGbb; + req.byte_bus_id = 3; + req.transaction_num = 9; + + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(ch->request(0, req, {}, resp, resp_payload)); + REQUIRE(saw_gbb); + REQUIRE(resp.acf_msg_type == acf::kAcfMsgTypeGbb); // make_response() copies acf_msg_type from req } TEST_CASE("shmem Channel::request answers Acknowledge by default when no handler is set", @@ -62,9 +143,13 @@ TEST_CASE("shmem Channel::request answers Acknowledge by default when no handler REQUIRE(acf::response_kind_of(resp) == acf::ResponseKind::Acknowledge); } -// ── Lifecycle ───────────────────────────────────────────────────────────────── +// ── Lifecycle (REQ-SHMEM-005/010: Channel::request() plays both the "send" +// and "recv" role of c-RCP's own shmem_side_send()/shmem_side_recv(), each +// of which reports RCP_ERR_CLOSED once its side is closed — shmem.c:64-67/ +// 100-113,109-113) ────────────────────────────────────────────────────── -TEST_CASE("shmem Channel::request returns ErrClosed after close()", "[shmem][REQ-SHMEM-003]") { +TEST_CASE("shmem Channel::request returns ErrClosed after close()", + "[shmem][REQ-SHMEM-005][REQ-SHMEM-010]") { auto ch = new_channel(1); ch->close(); @@ -74,40 +159,204 @@ TEST_CASE("shmem Channel::request returns ErrClosed after close()", "[shmem][REQ REQUIRE(ch->request(0, req, {}, resp, resp_payload) == ErrClosed); } -// ── Registry ────────────────────────────────────────────────────────────────── +// ── Capacity bound / backpressure (this pass's core fix: no bound existed +// at all before) ───────────────────────────────────────────────────────── -TEST_CASE("shmem Registry::lookup finds a channel registered via add_channel", - "[shmem][REQ-SHMEM-004]") { - auto reg = new_registry(); - auto ch = new_channel(7); - REQUIRE_FALSE(reg->add_channel(ch)); - - std::shared_ptr out; - REQUIRE_FALSE(reg->lookup(7, out)); - REQUIRE(out == ch); +TEST_CASE("shmem Channel::queue_capacity reflects the constructor argument, clamped into " + "[1, kMaxQueueCapacity]", + "[shmem][REQ-SHMEM-006]") { + REQUIRE(new_channel(1, /*queue_capacity=*/4)->queue_capacity() == 4); + // 0 is clamped up to 1 -- a 0-slot pool would make every request() call + // unconditionally busy, never a reachable, useful configuration (see + // rcp/shmem.hpp's FrameSlots::set_logical_capacity(), mirroring c-RCP's + // own rcp_shmem_avtp_pair_new() clamp, shmem.c:204). + REQUIRE(new_channel(1, /*queue_capacity=*/0)->queue_capacity() == 1); + // Clamped down to the physical ceiling. + REQUIRE(new_channel(1, /*queue_capacity=*/1000)->queue_capacity() == kMaxQueueCapacity); + // new_channel()'s own default matches Channel's own default. + REQUIRE(new_channel(1)->queue_capacity() == kDefaultQueueCapacity); } -TEST_CASE("shmem Registry::close closes all registered channels", "[shmem][REQ-SHMEM-005]") { - auto reg = new_registry(); - auto ch = new_channel(7); - REQUIRE_FALSE(reg->add_channel(ch)); - REQUIRE_FALSE(reg->close()); - - std::shared_ptr out; - REQUIRE(reg->lookup(7, out) == ErrClosed); - REQUIRE_FALSE(ch->ok()); // the channel itself was closed too, not just delisted +TEST_CASE("shmem Channel::queue_depth reports one occupied slot while a request is being " + "handled, and zero once it returns", + "[shmem][REQ-SHMEM-006]") { + auto ch = new_channel(1, /*queue_capacity=*/4); + size_t depth_seen_by_handler = 999; + ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + depth_seen_by_handler = ch->queue_depth(); + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); + REQUIRE(ch->queue_depth() == 0); auto req = standard_request(1, 1); acf::AcfMessageInfo resp; std::vector resp_payload; - REQUIRE(ch->request(0, req, {}, resp, resp_payload) == ErrClosed); + REQUIRE_FALSE(ch->request(0, req, {}, resp, resp_payload)); + + REQUIRE(depth_seen_by_handler == 1); + REQUIRE(ch->queue_depth() == 0); // slot released once request() returns +} + +TEST_CASE("shmem Channel::request returns ErrBusy immediately, without blocking, once " + "queue_capacity requests are already in flight", + "[shmem][REQ-SHMEM-006]") { + auto ch = new_channel(1, /*queue_capacity=*/1); + + std::mutex mu; + std::condition_variable release_cv; + bool entered = false; + bool may_release = false; + + ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + { + std::lock_guard lk(mu); + entered = true; + } + release_cv.notify_all(); + std::unique_lock lk(mu); + release_cv.wait(lk, [&] { return may_release; }); + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); + + auto req1 = standard_request(1, 1); + acf::AcfMessageInfo resp1; + std::vector resp1_payload; + std::error_code ec1; + std::thread blocked_caller([&] { ec1 = ch->request(0, req1, {}, resp1, resp1_payload); }); + + { + std::unique_lock lk(mu); + release_cv.wait(lk, [&] { return entered; }); + } + + // The one and only slot is occupied by blocked_caller's still-running + // handler -- a second, concurrent caller must be rejected immediately + // (REQ-SHMEM-006), not queued and not blocked waiting for room. + auto req2 = standard_request(1, 2); + acf::AcfMessageInfo resp2; + std::vector resp2_payload; + REQUIRE(ch->request(1, req2, {}, resp2, resp2_payload) == ErrBusy); + + { + std::lock_guard lk(mu); + may_release = true; + } + release_cv.notify_all(); + blocked_caller.join(); + + REQUIRE_FALSE(ec1); + REQUIRE(acf::response_kind_of(resp1) == acf::ResponseKind::Acknowledge); + + // The slot is free again -- a request that would previously have been + // rejected now succeeds. + acf::AcfMessageInfo resp3; + std::vector resp3_payload; + REQUIRE_FALSE(ch->request(2, req2, {}, resp3, resp3_payload)); +} + +TEST_CASE("shmem Channel::request admits up to queue_capacity concurrent callers and rejects " + "only the overflow", + "[shmem][REQ-SHMEM-006]") { + constexpr size_t kCapacity = 3; + auto ch = new_channel(1, kCapacity); + + std::mutex mu; + std::condition_variable entered_cv, release_cv; + size_t entered_count = 0; + bool may_release = false; + + ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + { + std::lock_guard lk(mu); + ++entered_count; + } + entered_cv.notify_all(); + std::unique_lock lk(mu); + release_cv.wait(lk, [&] { return may_release; }); + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); + + std::vector threads; + std::atomic ok_count{0}; + for (size_t i = 0; i < kCapacity; ++i) { + threads.emplace_back([&, i] { + auto req = standard_request(1, static_cast(i)); + acf::AcfMessageInfo resp; + std::vector resp_payload; + if (!ch->request(i, req, {}, resp, resp_payload)) ok_count.fetch_add(1); + }); + } + + { + std::unique_lock lk(mu); + entered_cv.wait(lk, [&] { return entered_count == kCapacity; }); + } + + // Every slot is now occupied -- one more concurrent caller must see + // ErrBusy while all kCapacity handlers are still blocked. + auto overflow_req = standard_request(1, 99); + acf::AcfMessageInfo overflow_resp; + std::vector overflow_resp_payload; + REQUIRE(ch->request(kCapacity, overflow_req, {}, overflow_resp, overflow_resp_payload) == + ErrBusy); + + { + std::lock_guard lk(mu); + may_release = true; + } + release_cv.notify_all(); + for (auto& t : threads) t.join(); + + REQUIRE(ok_count.load() == static_cast(kCapacity)); } -// ── Concurrency ─────────────────────────────────────────────────────────────── +// ── No cross-talk between concurrent in-flight requests (REQ-SHMEM-002/003 +// adapted: c-RCP's two directions are independent and never observe each +// other's frames -- Channel::request() has one direction per call, so the +// analogous property is that one caller's own request/response bytes are +// never delivered to, or returned from, a different concurrent caller) ── + +TEST_CASE("shmem Channel::request never delivers one concurrent caller's payload to another", + "[shmem][REQ-SHMEM-002][REQ-SHMEM-003]") { + auto ch = new_channel(1, /*queue_capacity=*/8); + ch->set_handler([](size_t, const acf::AcfMessageInfo& req, const std::vector& payload, + acf::AcfMessageInfo& out_resp, std::vector& out_payload) { + out_resp = acf::make_response(req, acf::ResponseKind::ReadResponse); + out_payload = payload; + return std::error_code{}; + }); + + constexpr int kThreads = 8; + std::vector threads; + std::atomic mismatches{0}; + for (int t = 0; t < kThreads; ++t) { + threads.emplace_back([&, t] { + for (int i = 0; i < 20; ++i) { + auto req = standard_request(1, static_cast(t)); + std::vector payload{static_cast(t), static_cast(i)}; + acf::AcfMessageInfo resp; + std::vector resp_payload; + if (ch->request(static_cast(t), req, payload, resp, resp_payload)) continue; + if (resp_payload != payload) mismatches.fetch_add(1); + } + }); + } + for (auto& th : threads) th.join(); + + REQUIRE(mismatches.load() == 0); +} + +// ── Concurrency (unchanged from before this pass) ─────────────────────────── TEST_CASE("shmem Channel::request is thread-safe for concurrent callers", "[shmem][REQ-SHMEM-006]") { - auto ch = new_channel(1); + auto ch = new_channel(1, /*queue_capacity=*/8); std::atomic call_count{0}; ch->set_handler([&](size_t, const acf::AcfMessageInfo& req, const std::vector&, acf::AcfMessageInfo& out_resp, std::vector&) { @@ -138,7 +387,7 @@ TEST_CASE("shmem Channel::request is thread-safe for concurrent callers", // ── Per-call client id pass-through ────────────────────────────────────────── TEST_CASE("shmem Channel::request forwards the caller-supplied client id to the handler unchanged", - "[shmem][REQ-SHMEM-007]") { + "[shmem][REQ-SHMEM-001]") { auto ch = new_channel(1); size_t seen = 999; ch->set_handler([&](size_t client, const acf::AcfMessageInfo& req, @@ -158,7 +407,7 @@ TEST_CASE("shmem Channel::request forwards the caller-supplied client id to the // ── Idempotent close ────────────────────────────────────────────────────────── -TEST_CASE("shmem Channel::close and Registry::close are idempotent", "[shmem][REQ-SHMEM-008]") { +TEST_CASE("shmem Channel::close and Registry::close are idempotent", "[shmem][registry]") { auto ch = new_channel(1); ch->close(); ch->close(); // second call must not crash @@ -168,3 +417,110 @@ TEST_CASE("shmem Channel::close and Registry::close are idempotent", "[shmem][RE REQUIRE_FALSE(reg->close()); REQUIRE_FALSE(reg->close()); // second call must not crash } + +// ── shared_ptr ownership (REQ-SHMEM-008/009 adapted: c-RCP's own +// refcounted rcp_avtp_transport_t sides guarantee releasing one holder +// neither invalidates another holder's own use of the shared state +// (shmem.c:151-181) nor frees it more than once, regardless of release +// order -- std::shared_ptr already gives cpp-RCP both properties +// automatically, see rcp/shmem.hpp's own Channel class comment) ────────── + +TEST_CASE("shmem a Channel remains usable through one shared_ptr holder after another holder " + "of the same Channel is released", + "[shmem][REQ-SHMEM-008]") { + auto reg = new_registry(); + auto ch = new_channel(3); + REQUIRE_FALSE(reg->add_channel(ch)); // Registry now holds its own shared_ptr copy too + + ch->set_handler([](size_t, const acf::AcfMessageInfo& req, const std::vector&, + acf::AcfMessageInfo& out_resp, std::vector&) { + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); + + std::shared_ptr from_registry; + REQUIRE_FALSE(reg->lookup(3, from_registry)); + ch.reset(); // this test's own local holder released; Registry's copy remains + + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE_FALSE(from_registry->request(0, req, {}, resp, resp_payload)); + REQUIRE(acf::response_kind_of(resp) == acf::ResponseKind::Acknowledge); +} + +TEST_CASE("shmem a Channel's state is freed exactly once regardless of shared_ptr release order", + "[shmem][REQ-SHMEM-009]") { + // No public way to observe the underlying allocation's lifetime + // directly; this pins the externally-visible contract instead (the + // same "no in-test assertion, verified by this suite's own sanitizer + // run" approach c-RCP's own equivalent test documents, tests/ + // test_shmem.c's test_pair_state_freed_after_releasing_both_sides) -- + // constructing, copying, and releasing shared_ptr handles in + // either order must neither crash nor leak. + std::weak_ptr weak; + { + auto a = new_channel(5); + weak = a; + auto b = a; // second holder + a.reset(); + REQUIRE_FALSE(weak.expired()); // b still holds it + b.reset(); + } + REQUIRE(weak.expired()); // freed once both holders released, either order +} + +// ── Registry ────────────────────────────────────────────────────────────────── +// Registry itself has no c-RCP analog at all (rcp_shmem_avtp_pair_new() +// returns exactly one pair, with no keyed lookup of any kind) -- these +// stay untagged with a REQ-SHMEM-* id rather than force-fitting one. + +TEST_CASE("shmem Registry::lookup finds a channel registered via add_channel", + "[shmem][registry]") { + auto reg = new_registry(); + auto ch = new_channel(7); + REQUIRE_FALSE(reg->add_channel(ch)); + + std::shared_ptr out; + REQUIRE_FALSE(reg->lookup(7, out)); + REQUIRE(out == ch); +} + +TEST_CASE("shmem Registry::add_channel rejects a duplicate stream_key", "[shmem][registry]") { + auto reg = new_registry(); + REQUIRE_FALSE(reg->add_channel(new_channel(7))); + REQUIRE(reg->add_channel(new_channel(7)) == ErrAlreadyExists); +} + +TEST_CASE("shmem Registry::deregister removes and closes a channel", "[shmem][registry]") { + auto reg = new_registry(); + auto ch = new_channel(7); + REQUIRE_FALSE(reg->add_channel(ch)); + REQUIRE_FALSE(reg->deregister(7)); + + std::shared_ptr out; + REQUIRE(reg->lookup(7, out) == ErrNotFound); + REQUIRE_FALSE(ch->ok()); +} + +TEST_CASE("shmem Registry::close closes all registered channels", "[shmem][registry]") { + auto reg = new_registry(); + auto ch = new_channel(7); + REQUIRE_FALSE(reg->add_channel(ch)); + REQUIRE_FALSE(reg->close()); + + std::shared_ptr out; + REQUIRE(reg->lookup(7, out) == ErrClosed); + REQUIRE_FALSE(ch->ok()); // the channel itself was closed too, not just delisted + + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + REQUIRE(ch->request(0, req, {}, resp, resp_payload) == ErrClosed); +} + +TEST_CASE("shmem Registry::channels enumerates every registered channel", "[shmem][registry]") { + auto reg = new_registry(); + for (uint64_t k : {1, 2, 3}) REQUIRE_FALSE(reg->add_channel(new_channel(k))); + REQUIRE(reg->channels().size() == 3); +} From a77a898722dcd0fd0ce086f386bf2e71bdf596fc Mon Sep 17 00:00:00 2001 From: Matt <47545907+SoundMatt@users.noreply.github.com> Date: Sat, 22 Aug 2026 02:47:39 -0700 Subject: [PATCH 2/2] =?UTF-8?q?rewrite(phase5):=20admin.hpp=20=E2=80=94=20?= =?UTF-8?q?fixed-capacity=20subscriber/counter=20bounds,=20port=20deadlock?= =?UTF-8?q?=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of the v3.0.0 ground-up rewrite (cpp-RCP #129, ROADMAP.md Phase 17), branch rewrite/v3-from-c-rcp. Phase 5 (transport layer), final module. ## Ported (c-RCP src/admin.c 247 lines / include/rcp/admin.h 177 lines) - admin.c:109-110 (RCP_ADMIN_MAX_SUBSCRIBERS=16, RCP_ADMIN_MAX_COUNTERS=256) -> std::array<..., kMaxSubscribers>/std::array<..., kMaxCounters> bounds on Server's subscriber list and counter table, replacing cpp-RCP's unbounded std::vector/std::map. Matches this rewrite's fixed-capacity, no-dynamic-allocation-on-safety-relevant-paths architecture (already applied to server.hpp's admission queue and shmem.hpp's FrameSlots). - admin.c:133-147 (rcp_admin_server_subscribe, capacity check before add) -> Server::subscribe() returns AdminErrc::subscriber_capacity_exceeded once kMaxSubscribers is reached, matching c-RCP's fail-closed "cb not added" behavior (admin.h:145) rather than growing unbounded. - admin.c:173-198 (rcp_admin_server_record_counter, capacity check on a genuinely new counter name+labels pair) -> counter_capacity_exceeded once kMaxCounters distinct counters exist (admin.h:158), existing counters still update past the bound. - Ported c-RCP's two boundary tests (subscribe-at-capacity, record_counter-at-capacity) into Catch2 TEST_CASEs. ## Bug found and fixed (real concurrency bug, same class as shmem.hpp/#153) emit() (ported from rcp_admin_server_emit(), admin.c:149-171) held the subscriber-list mutex across every subscriber callback invocation -- a subscriber callback that itself calls back into Server (a realistic pattern for a metrics/event sink) would deadlock. Fixed by copying the subscriber list out under the lock, then invoking callbacks unlocked -- same fix shape as shmem.hpp's Channel::request() handler-invocation fix in #153. ## Deliberate non-port, documented in-file c-RCP's admin.c is caller-driven: rcp_registry_t was retired project- wide (per c-RCP's own history) and every admin_server_t call site now passes state in directly rather than looking it up from a live registry. cpp-RCP's admin.hpp instead holds a reference to shmem::Registry and looks up endpoint/channel state through it live. NOT migrated to c-RCP's caller-driven pattern: c-RCP's rationale for going caller-driven was retiring rcp_registry_t entirely, which doesn't apply here -- shmem::Registry is live, current, and still the single source of truth for channel state in this rewrite (see shmem.hpp, #153). Reasoning recorded in admin.hpp's own header comment, not just here. ## Verification include/rcp/admin.hpp: 128 -> 338 lines. tests/test_admin.cpp: 123 -> 200 lines. Full tree rebuilds clean (0 errors, 0 warnings) from a from-scratch rebuild against the fully-merged rewrite/v3-from-c-rcp base (picked up #151/#152/#153 via merge-forward first), 58/58 ctest suites pass (100%) -- independently re-verified, not just the porting agent's own report. ## Phase 5 status udp (#151), l2 (#152), shmem (#153) already merged; this closes out admin, the fourth and final "port from c-RCP" module in Phase 5's scope. tls/capi remain as re-verify-only per ROADMAP.md Phase 17 — confirmed clean against the new core, no content changes needed. Co-Authored-By: Claude Sonnet 5 Signed-off-by: Matt <47545907+SoundMatt@users.noreply.github.com> --- include/rcp/admin.hpp | 276 +++++++++++++++++++++++++++++++++++++----- tests/test_admin.cpp | 77 ++++++++++++ 2 files changed, 320 insertions(+), 33 deletions(-) diff --git a/include/rcp/admin.hpp b/include/rcp/admin.hpp index 04b7c0d..fd740bf 100644 --- a/include/rcp/admin.hpp +++ b/include/rcp/admin.hpp @@ -7,36 +7,186 @@ // fusa:req REQ-ADMIN-007 // fusa:req REQ-ADMIN-008 -// In-process Admin API: stream listing, SSE events, Prometheus metrics (v0.26.0). +// In-process Admin API: stream listing, SSE-style events, Prometheus +// metrics (Phase 17 ground-up rewrite pass, cpp-RCP issue #129 — ROADMAP.md +// Phase 17 §5, "Transport: udp, l2, shmem, admin"). // // AdminServer is a lightweight in-process interface: callers can query -// which streams are registered, subscribe to events (SSE-style push +// which streams are registered (via shmem::Registry — see "Registry +// reference vs. caller-driven" below), subscribe to events (SSE-style push // channel), and snapshot Prometheus-format text metrics. An actual HTTP // binding is out of scope; use a libmicrohttpd or Asio adapter to expose // the HTTP surface. // -// Rebound (cpp-RCP-FS-03, #86): this used to report `rcp::Registry`'s -// registered Zones. `Zone`/`rcp::Registry` are retired (cpp-RCP-FS-01, #84); -// AdminServer is now bound to `rcp::shmem::Registry` (rcp/shmem.hpp), the -// TC18-shaped in-process registry keyed by opaque stream_key, and reports -// streams instead of zones. +// ── What ported from c-RCP's admin.h/admin.c, and what deliberately didn't ── +// c-RCP's admin.h/admin.c (include/rcp/admin.h, src/admin.c) is this pass's +// content source of truth for two genuine deltas ported below, plus one +// deliberate non-port: +// +// 1. emit() deadlock risk (PORTED). c-RCP's own admin.h header comment +// (lines ~34-42) flags this by name: "Deviation from cpp-RCP...: +// cpp-RCP's emit() holds its mutex for the duration of every subscriber +// callback invocation. This port invokes callbacks outside the lock +// instead...a subscriber that calls back into the same server...cannot +// deadlock." That described a real, live bug in this file before this +// pass: emit() held std::lock_guard lk(mu_) across the +// entire `for (auto& cb : subscribers_) cb(ev)` loop, so a subscriber +// callback that itself called srv.subscribe()/emit()/record_counter() +// (all of which also lock mu_, a plain non-recursive std::mutex) would +// deadlock against itself. Fixed below by copying the subscriber list +// out under the lock, releasing the lock, and only then invoking the +// callbacks — the same "copy under lock, invoke unlocked" idiom +// rcp/shmem.hpp's own Channel::request() already established in this +// codebase's sibling Phase 17 wave (copies handler_ under mu_, invokes +// the copy outside the lock — see that function's own comment) and +// c-RCP's admin.c itself uses (a fixed-size local[RCP_ADMIN_MAX_ +// SUBSCRIBERS] stack snapshot, admin.c's rcp_admin_server_emit()). +// Regression-tested below ("emit: a subscriber that calls back into the +// same server does not deadlock"). +// +// 2. Fixed-capacity subscribers_/counters_ (PORTED). c-RCP's +// RCP_ADMIN_MAX_SUBSCRIBERS (16) / RCP_ADMIN_MAX_COUNTERS (256) bound +// admin.c's own subscriber list and counter table — fixed embedded +// arrays, not heap allocations growable without bound (c-RCP admin.h's +// own [c-RCP-17] doc comment). Before this pass, this file's +// subscribers_/counters_ were an unbounded std::vector/ +// std::unordered_map — the same "no fixed-capacity/no-dynamic- +// allocation architecture" gap ROADMAP.md's Phase 17 introduction names +// as one motivation for this whole rewrite, and the same shape +// rcp/watchdog.hpp's Manager::kMaxStreams/kMaxCallbacks (also 16) fixed +// for this codebase's Phase 14 wave. Ported unchanged (16/256) below as +// kMaxSubscribers/kMaxCounters, backing std::array storage with a manual +// length counter — matching rcp/watchdog.hpp's Manager idiom exactly +// rather than introducing yet another generic bounded-container +// abstraction (rcp/request.hpp's detail::BoundedVector is +// deliberately scoped private to that file — this codebase's existing +// convention is each module rolling its own fixed std::array + _len_ +// pair, not sharing one container type across modules; see +// rcp/alloc.hpp's header comment for the same "each stateful primitive +// is its own small instance-owned thing" preference stated explicitly). +// subscribe()/record_counter() now return std::error_code +// (AdminErrc::subscriber_capacity_exceeded/counter_capacity_exceeded) +// instead of silently succeeding forever, mirroring +// rcp::watchdog::Manager::subscribe()'s identical return-shape +// precedent. Boundary tests ported below: "subscribe: at max succeeds, +// next fails" / "record_counter: at max succeeds, a repeat of an +// existing counter still succeeds, a genuinely new one fails" — the +// cpp-RCP equivalents of c-RCP's own +// test_subscribe_at_max_succeeds_then_next_fails/ +// test_record_counter_at_max_succeeds_then_next_new_one_fails +// (tests/test_admin.c). +// +// RCP_ADMIN_MAX_ENDPOINTS (64) deliberately does NOT transfer: it bounds +// c-RCP's own endpoints[] array, which exists only because c-RCP's +// admin.c tracks its registered-endpoint set itself (see point 3 below +// for why). This file's streams() has never tracked its own stream set +// — it reads live off shmem::Registry (rcp/shmem.hpp), whose own +// capacity (currently unbounded, a std::map — see that file's own +// Registry class comment, unchanged by this pass) is that file's +// decision to make, not this one's. +// +// 3. Registry reference vs. c-RCP's caller-driven register_endpoint()/ +// deregister_endpoint() (NOT PORTED — deliberate). c-RCP's own admin.h +// header comment (lines ~10-30) explains why ITS admin module dropped +// its own registry dependency entirely: c-RCP's admin.c used to wrap +// rcp.h's rcp_registry_t and enumerate its rcp_controller_t instances, +// but "rcp_registry_t has no TC18 counterpart -- there is no single +// generic registry of 'every controller' left to introspect" once that +// retired core was replaced, so c-RCP's admin.c became caller-driven +// instead: whatever code discovers or configures endpoints tells +// admin.c directly via rcp_admin_server_register_endpoint()/ +// _deregister_endpoint() (which just insert/remove a caller-supplied +// rcp_avtp_addr_t into admin.c's own fixed endpoints[] array — see +// point 2 above), and rcp_admin_server_endpoints() reports back exactly +// that caller-maintained membership. +// +// That rationale does not apply here, and porting it anyway would +// actively regress this codebase: unlike c-RCP's rcp_registry_t, +// rcp::shmem::Registry is NOT retired — it is a live, current, still- +// actively-maintained concept (rebuilt in this exact same Phase 17 wave, +// the sibling `phase5/shmem` branch this branch is based on), with a +// real keyed add_channel()/deregister()/lookup()/channels()/close() +// surface AdminServer::streams() already depends on. rcp/shmem.hpp's own +// header comment (its "Registry" class comment) explicitly commits to +// this: "rcp::admin::AdminServer (rcp/admin.hpp) ... only ever depend[s] +// on Registry's keyed add/lookup/enumerate surface ... never on +// Channel::request()'s internals, and that surface is unchanged here — +// verified by reading both files and their tests ... in full; neither +// calls Channel::request() at all." Migrating streams() to a caller- +// driven register_endpoint()/deregister_endpoint() pair would make +// AdminServer track a second, independent copy of "what streams exist" +// that callers would have to remember to keep in sync with Registry by +// hand (register on add_channel(), deregister on deregister()) — +// reintroducing exactly the dual-source-of-truth bug class a registry- +// reference design avoids by construction, for no offsetting benefit, +// since (unlike c-RCP) there is still a live, single, authoritative +// registry object to read from directly. Also verified this is not +// disproportionately load-bearing to keep either way: grepped this tree +// codebase-wide for AdminServer/rcp::admin/admin:: — no caller other +// than tests/test_admin.cpp constructs an AdminServer at all +// (rcp/config.hpp, in particular, has never wired one). Decision: +// AdminServer keeps its shmem::Registry& constructor reference and +// streams() keeps reading reg_.channels() live; StreamInfo/EventType/ +// Event/stream_key naming is otherwise unchanged from this file's +// pre-Phase-17 (v2.19.0) shape. +// +// Every other REQ-ADMIN-* citation and behavior (streams()/subscribe()/ +// emit()/record_counter()/metrics_text()) is unchanged in shape from the +// v2.19.0 rebind — see git history for that pass's own rationale +// (ROADMAP.md's "Retired-model residue cleanup" milestone). #pragma once #include "rcp.hpp" #include "shmem.hpp" +#include #include +#include #include #include #include #include #include -#include +#include #include namespace rcp { namespace admin { +// ── Errors ──────────────────────────────────────────────────────────────────── + +enum class AdminErrc : int { + // [Phase 17 c-RCP-reference pass, cpp-RCP issue #129] Ported from + // c-RCP's RCP_ADMIN_MAX_SUBSCRIBERS/RCP_ADMIN_MAX_COUNTERS fixed- + // capacity bound (admin.h) — see AdminServer::kMaxSubscribers/ + // kMaxCounters's own doc comment below and this file's header comment, + // delta #2. + subscriber_capacity_exceeded = 1, // AdminServer already holds kMaxSubscribers subscribers + counter_capacity_exceeded = 2, // AdminServer already tracks kMaxCounters distinct (name, labels) counters +}; + +inline const std::error_category& admin_category() noexcept { + struct Cat : std::error_category { + const char* name() const noexcept override { return "rcp.admin"; } + std::string message(int ev) const override { + switch (static_cast(ev)) { + case AdminErrc::subscriber_capacity_exceeded: + return "rcp/admin: AdminServer already holds its fixed maximum number of subscribers"; + case AdminErrc::counter_capacity_exceeded: + return "rcp/admin: AdminServer already tracks its fixed maximum number of counters"; + default: + return "rcp/admin: unknown error"; + } + } + }; + static Cat instance; + return instance; +} + +inline std::error_code make_error_code(AdminErrc e) noexcept { + return {static_cast(e), admin_category()}; +} + // ── StreamInfo ──────────────────────────────────────────────────────────────── struct StreamInfo { @@ -69,11 +219,23 @@ struct Counter { class AdminServer { public: + // [Phase 17 c-RCP-reference pass, cpp-RCP issue #129] Ported from + // c-RCP's include/rcp/admin.h RCP_ADMIN_MAX_SUBSCRIBERS/ + // RCP_ADMIN_MAX_COUNTERS (16/256 respectively) — see this file's header + // comment, delta #2, for the full rationale and for why + // RCP_ADMIN_MAX_ENDPOINTS does NOT have a counterpart here. Backs the + // fixed std::array storage below (subscribers_, counters_), not + // std::vector/std::unordered_map growable without bound. + static constexpr size_t kMaxSubscribers = 16; + static constexpr size_t kMaxCounters = 256; + explicit AdminServer(shmem::Registry& reg) : reg_(reg) {} - // streams returns a snapshot of all registered streams. + // streams returns a snapshot of all registered streams — read live off + // reg_ (shmem::Registry), not a copy AdminServer maintains itself; see + // this file's header comment, delta #3. std::vector streams() const { - auto channels = reg_.channels(); + auto channels = reg_.channels(); std::vector out; out.reserve(channels.size()); for (auto& ch : channels) { @@ -82,47 +244,95 @@ class AdminServer { return out; } - // subscribe registers an SSE callback for registry events. - void subscribe(EventCallback cb) { + // subscribe registers cb to be invoked (in registration order) on every + // subsequent emit() call. Returns AdminErrc::subscriber_capacity_ + // exceeded, unchanged, once kMaxSubscribers subscribers are already + // registered — see kMaxSubscribers's own doc comment above. + std::error_code subscribe(EventCallback cb) { std::lock_guard lk(mu_); - subscribers_.push_back(std::move(cb)); + if (subscribers_len_ >= kMaxSubscribers) return make_error_code(AdminErrc::subscriber_capacity_exceeded); + subscribers_[subscribers_len_] = std::move(cb); + ++subscribers_len_; + return {}; } - // emit broadcasts an event to all subscribers. + // emit broadcasts ev to every registered subscriber, in registration + // order. The subscriber list is copied out under mu_ and every callback + // is invoked AFTER mu_ is released — see this file's header comment, + // delta #1: a subscriber that calls back into this same AdminServer + // (subscribe()/emit()/record_counter()) cannot deadlock. void emit(Event ev) { - std::lock_guard lk(mu_); - for (auto& cb : subscribers_) cb(ev); + std::array local; + size_t n; + { + std::lock_guard lk(mu_); + n = subscribers_len_; + for (size_t i = 0; i < n; ++i) local[i] = subscribers_[i]; + } + // Invoked outside the lock — see the deviation note above. + for (size_t i = 0; i < n; ++i) local[i](ev); } - // record_counter increments a named metric counter. - void record_counter(const std::string& name, const std::string& labels, double delta) { + // record_counter adds delta to the running total of the counter + // identified by (name, labels) — a distinct running total is kept per + // unique (name, labels) pair. Returns AdminErrc::counter_capacity_ + // exceeded without recording delta if (name, labels) is not already + // tracked and this AdminServer already holds kMaxCounters distinct + // counters; an already-tracked (name, labels) pair always succeeds + // regardless of how many other counters exist — see kMaxCounters's own + // doc comment above. + std::error_code record_counter(const std::string& name, const std::string& labels, double delta) { std::lock_guard lk(mu_); - auto& c = counters_[name + "{" + labels + "}"]; - c.name = name; - c.labels = labels; - c.value += delta; + for (size_t i = 0; i < counters_len_; ++i) { + if (counters_[i].name == name && counters_[i].labels == labels) { + counters_[i].value += delta; + return {}; + } + } + if (counters_len_ >= kMaxCounters) return make_error_code(AdminErrc::counter_capacity_exceeded); + counters_[counters_len_] = Counter{name, labels, delta}; + ++counters_len_; + return {}; } // metrics_text returns Prometheus text-format metric lines. std::string metrics_text() const { std::lock_guard lk(mu_); - std::ostringstream oss; - for (auto& kv : counters_) { - oss << "# TYPE " << kv.second.name << " counter\n"; - oss << kv.second.name; - if (!kv.second.labels.empty()) - oss << "{" << kv.second.labels << "}"; - oss << " " << kv.second.value << "\n"; + std::ostringstream oss; + for (size_t i = 0; i < counters_len_; ++i) { + oss << "# TYPE " << counters_[i].name << " counter\n"; + oss << counters_[i].name; + if (!counters_[i].labels.empty()) oss << "{" << counters_[i].labels << "}"; + oss << " " << counters_[i].value << "\n"; } return oss.str(); } + // Introspection for tests, not part of the subscribe()/record_counter() + // contract itself — always <= kMaxSubscribers/kMaxCounters. + size_t subscriber_count() const noexcept { + std::lock_guard lk(mu_); + return subscribers_len_; + } + size_t counter_count() const noexcept { + std::lock_guard lk(mu_); + return counters_len_; + } + private: - shmem::Registry& reg_; - mutable std::mutex mu_; - std::vector subscribers_; - std::unordered_map counters_; + shmem::Registry& reg_; + mutable std::mutex mu_; // protects subscribers_/subscribers_len_/counters_/counters_len_ + std::array subscribers_{}; + size_t subscribers_len_ = 0; // always <= kMaxSubscribers + std::array counters_{}; + size_t counters_len_ = 0; // always <= kMaxCounters }; } // namespace admin } // namespace rcp + +// Enable std::error_code construction from rcp::admin::AdminErrc. +namespace std { +template <> +struct is_error_code_enum : true_type {}; +} // namespace std diff --git a/tests/test_admin.cpp b/tests/test_admin.cpp index 9aea901..d2082ea 100644 --- a/tests/test_admin.cpp +++ b/tests/test_admin.cpp @@ -121,3 +121,80 @@ TEST_CASE("admin: concurrent record_counter and emit are thread-safe", REQUIRE(text.find("rcp.commands.total") != std::string::npos); REQUIRE(text.find(std::to_string(kThreads * kPerThread)) != std::string::npos); } + +// ── [Phase 17 c-RCP-reference pass, cpp-RCP issue #129] emit() deadlock fix ── +// See rcp/admin.hpp's own header comment, delta #1: emit() used to hold mu_ +// across every subscriber callback invocation, so a subscriber calling back +// into the same AdminServer (subscribe()/emit()/record_counter()) would +// deadlock against a plain non-recursive std::mutex. This test hangs forever +// pre-fix and completes immediately post-fix — a real regression test, not +// just a documentation stand-in. +TEST_CASE("admin: emit — a subscriber that calls back into the same server does not deadlock", + "[admin][REQ-ADMIN-004]") { + shmem::Registry reg; + admin::AdminServer srv(reg); + + int reentrant_calls = 0; + bool reentered = false; + srv.subscribe([&](const admin::Event& ev) { + // Re-enter the server from inside the callback exactly once, on + // every distinct kind of call emit()'s own mu_ also guards, to + // exercise all three re-entrant paths the header comment names. + if (!reentered) { + reentered = true; + REQUIRE_FALSE(srv.record_counter("rcp.reentrant.total", "", 1.0)); + REQUIRE_FALSE(srv.subscribe([&](const admin::Event&) { ++reentrant_calls; })); + srv.emit({admin::EventType::StatusUpdate, ev.stream_key, {}}); + } + }); + + srv.emit({admin::EventType::StreamRegistered, 7, {}}); + + // Reaching here at all (rather than hanging) is the fix; the counts + // below additionally confirm the re-entrant calls actually took effect. + REQUIRE(srv.counter_count() == 1); + REQUIRE(srv.subscriber_count() == 2); + REQUIRE(reentrant_calls == 1); +} + +// ── [c-RCP-17] Fixed-capacity subscriber list / counter table ─────────────── +// Boundary coverage ported from c-RCP's tests/test_admin.c +// test_subscribe_at_max_succeeds_then_next_fails/ +// test_record_counter_at_max_succeeds_then_next_new_one_fails — see +// rcp/admin.hpp's own header comment, delta #2. + +TEST_CASE("admin: subscribe at max succeeds, then the next one fails", "[admin][REQ-ADMIN-003]") { + shmem::Registry reg; + admin::AdminServer srv(reg); + + for (size_t i = 0; i < admin::AdminServer::kMaxSubscribers; ++i) { + REQUIRE_FALSE(srv.subscribe([](const admin::Event&) {})); + } + REQUIRE(srv.subscriber_count() == admin::AdminServer::kMaxSubscribers); + + // One more, at capacity: rejected, not silently grown. + auto ec = srv.subscribe([](const admin::Event&) {}); + REQUIRE(ec == admin::AdminErrc::subscriber_capacity_exceeded); + REQUIRE(srv.subscriber_count() == admin::AdminServer::kMaxSubscribers); +} + +TEST_CASE("admin: record_counter at max succeeds, a repeat still succeeds, a new one fails", + "[admin][REQ-ADMIN-005]") { + shmem::Registry reg; + admin::AdminServer srv(reg); + + for (size_t i = 0; i < admin::AdminServer::kMaxCounters; ++i) { + REQUIRE_FALSE(srv.record_counter("counter_" + std::to_string(i), "", 1.0)); + } + REQUIRE(srv.counter_count() == admin::AdminServer::kMaxCounters); + + // A repeat delta against an already-tracked counter still succeeds at + // capacity -- only a genuinely new (name, labels) pair is rejected. + REQUIRE_FALSE(srv.record_counter("counter_0", "", 1.0)); + REQUIRE(srv.counter_count() == admin::AdminServer::kMaxCounters); + + // One more, genuinely new, at capacity: rejected, not silently grown. + auto ec = srv.record_counter("one_too_many", "", 1.0); + REQUIRE(ec == admin::AdminErrc::counter_capacity_exceeded); + REQUIRE(srv.counter_count() == admin::AdminServer::kMaxCounters); +}