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 e97b00d532dd5ec289a797b1267a090fd8bf6771 Mon Sep 17 00:00:00 2001 From: Matt <47545907+SoundMatt@users.noreply.github.com> Date: Sat, 22 Aug 2026 02:35:16 -0700 Subject: [PATCH 2/2] =?UTF-8?q?fix(shmem):=20correct=20REQ-SHMEM-004/007?= =?UTF-8?q?=20test=20coverage=20=E2=80=94=20numbering=20mix-up,=20not=20a?= =?UTF-8?q?=20real=20gap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior commit's header comment claimed REQ-SHMEM-004/007 had 'no analog' in the new architecture, based on c-RCP's own shmem.h numbering (where those ids mean recv() timeout / recv() destination-buffer-too- small). That was a numbering mix-up, not an actual gap: THIS project's .fusa-reqs.json assigns REQ-SHMEM-004 to 'Registry::lookup finds a channel registered via add_channel' and REQ-SHMEM-007 to 'the caller-supplied client id is forwarded to the handler unchanged' -- both ordinary, already-implemented, directly testable behaviors. - REQ-SHMEM-004: the existing 'Registry::lookup finds a channel registered via add_channel' test already covered this exact behavior -- it was just missing the requirement tag ([registry] only). Added the tag. - REQ-SHMEM-007: no test existed for client-id forwarding at all. Added one, asserting the handler sees exactly the client id passed to request() across two different values (not cached/defaulted from a prior call). Fixes the cpfusa trace (100%% coverage) gate failure on PR #153 (tested=410/total=412 -- these were the two untested entries). Co-Authored-By: Claude Sonnet 5 Signed-off-by: Matt <47545907+SoundMatt@users.noreply.github.com> --- tests/test_shmem.cpp | 55 +++++++++++++++++++++++++++++++++++--------- 1 file changed, 44 insertions(+), 11 deletions(-) diff --git a/tests/test_shmem.cpp b/tests/test_shmem.cpp index 01f4fe8..77f9f9a 100644 --- a/tests/test_shmem.cpp +++ b/tests/test_shmem.cpp @@ -1,21 +1,25 @@ // 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 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). +// (cpp-RCP issue #129 Phase 5 wave 2). +// +// NOTE: an earlier draft of this comment claimed REQ-SHMEM-004/007 had "no +// analog" here, based on c-RCP's own shmem.h numbering (where those ids mean +// recv() timeout / recv() destination-buffer-too-small). That was a +// numbering mix-up: THIS project's .fusa-reqs.json assigns REQ-SHMEM-004 to +// "Registry::lookup finds a channel registered via add_channel" and +// REQ-SHMEM-007 to "the caller-supplied client id is forwarded to the +// handler unchanged" — both ordinary, directly testable behaviors, covered +// below like every other entry in this file. #include #include @@ -132,6 +136,32 @@ TEST_CASE("shmem Channel::request round-trips an ACF_GBB request through the GBB REQUIRE(resp.acf_msg_type == acf::kAcfMsgTypeGbb); // make_response() copies acf_msg_type from req } +TEST_CASE("shmem Channel::request forwards the caller-supplied client id to the handler " + "unchanged", + "[shmem][REQ-SHMEM-007]") { + auto ch = new_channel(5); + size_t seen_client = static_cast(-1); + ch->set_handler([&](size_t client, const acf::AcfMessageInfo& req, + const std::vector&, acf::AcfMessageInfo& out_resp, + std::vector&) { + seen_client = client; + out_resp = acf::make_response(req, acf::ResponseKind::Acknowledge); + return std::error_code{}; + }); + + auto req = standard_request(1, 1); + acf::AcfMessageInfo resp; + std::vector resp_payload; + + REQUIRE_FALSE(ch->request(/*client=*/12345, req, {}, resp, resp_payload)); + REQUIRE(seen_client == 12345); + + // A different client id on a second call is forwarded independently -- + // not cached or defaulted from the first call. + REQUIRE_FALSE(ch->request(/*client=*/0, req, {}, resp, resp_payload)); + REQUIRE(seen_client == 0); +} + TEST_CASE("shmem Channel::request answers Acknowledge by default when no handler is set", "[shmem][REQ-SHMEM-002]") { auto ch = new_channel(1); @@ -472,11 +502,14 @@ TEST_CASE("shmem a Channel's state is freed exactly once regardless of shared_pt // ── 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. +// returns exactly one pair, with no keyed lookup of any kind) -- most of +// these stay untagged with a REQ-SHMEM-* id rather than force-fitting one. +// REQ-SHMEM-004 is the one exception: this project's own .fusa-reqs.json +// assigns that id specifically to the lookup-finds-a-registered-channel +// behavior below. TEST_CASE("shmem Registry::lookup finds a channel registered via add_channel", - "[shmem][registry]") { + "[shmem][registry][REQ-SHMEM-004]") { auto reg = new_registry(); auto ch = new_channel(7); REQUIRE_FALSE(reg->add_channel(ch));