diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6610537..5b3ffed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,9 +2,9 @@ name: CI on: push: - branches: [main] + branches: [main, devel] pull_request: - branches: [main] + branches: [main, devel] workflow_dispatch: jobs: diff --git a/rmw_unix_socket_cpp/CMakeLists.txt b/rmw_unix_socket_cpp/CMakeLists.txt index a8c2d34..079a6fb 100644 --- a/rmw_unix_socket_cpp/CMakeLists.txt +++ b/rmw_unix_socket_cpp/CMakeLists.txt @@ -207,6 +207,11 @@ if(BUILD_TESTING) target_link_libraries(test_rmw_qos ${_rmw_test_libs} ${_test_msg_deps}) target_include_directories(test_rmw_qos PRIVATE src) + # White-box: TL replay-cache pruning (deliberately inspects publisher internals) + ament_add_gtest(test_internal_publisher_cache test/test_internal_publisher_cache.cpp) + target_link_libraries(test_internal_publisher_cache ${_rmw_test_libs} ${_test_msg_deps}) + target_include_directories(test_internal_publisher_cache PRIVATE src) + # Integration: fork()-based two-process tests pushing >4 MB payloads across # a real PID boundary through the shared-memory path. ament_add_gtest(test_rmw_cross_process test/test_rmw_cross_process.cpp) diff --git a/rmw_unix_socket_cpp/DESIGN.md b/rmw_unix_socket_cpp/DESIGN.md index f818332..0381999 100644 --- a/rmw_unix_socket_cpp/DESIGN.md +++ b/rmw_unix_socket_cpp/DESIGN.md @@ -16,7 +16,7 @@ The middleware has four components: a shared-memory discovery registry, an AF_UN **Serialization (CDR via fastcdr).** Messages are serialized to CDR using `fastcdr` driven by the `rosidl_typesupport_fastrtps_cpp` callbacks generated for each message type. This is the same encoding the default DDS-based RMWs use. Because the serialize and deserialize routines are compiled per message type, there is no runtime field walking. The Serialization section explains why an earlier introspection-based serializer was abandoned. -**Wait mechanism (`epoll` + `eventfd`).** `rmw_wait()` blocks on `epoll`, watching the receive-socket file descriptors (fds) of every subscription, service, and client in the wait set plus an `eventfd` per guard condition. `epoll` is used here rather than `poll`/`select`; the wait section explains why. There are no background receiver threads: all socket draining happens inside `rmw_wait()`, which the executor already calls in a tight loop. This keeps the data path single-threaded per process and avoids lock contention on internal queues. +**Wait mechanism (`epoll` + `eventfd` + a doorbell).** `rmw_wait()` blocks on `epoll`, watching the receive-socket file descriptors (fds) of every subscription, service, and client in the wait set plus an `eventfd` per guard condition, plus one per-context doorbell socket that other processes ring after any registry mutation, so a blocked wait learns about graph changes without polling (see The doorbell under Wait set). `epoll` is used here rather than `poll`/`select`; the wait section explains why. There are no background receiver threads: all socket draining happens inside `rmw_wait()`, which the executor already calls in a tight loop. This keeps the data path single-threaded per process and avoids lock contention on internal queues. **End-to-end publish flow.** On `rmw_publish`, the node serializes the ROS message to CDR once — into a heap payload for small messages, or directly into the shared-memory ring record for large ones (see Staging and fanout). It fills in a fixed 37-byte wire header (sender GID, sequence number, send timestamp, payload size, and a type byte). It then reads the registry's `generation` counter and compares it to the value cached on the publisher. If the graph has not changed, the publisher reuses its cached list of subscriber socket paths and never touches the registry. Only when `generation` has moved does it re-scan the registry to rebuild that list. For each subscriber path it issues one `sendmsg` with gather I/O, so the header and payload become one datagram with no intermediate copy. Sends are non-blocking and best-effort, one kernel copy into each subscriber's socket buffer. Payloads of 64 KiB and above take a different route: the publisher stages the bytes once in its shared-memory ring and each subscriber receives only a 32-byte descriptor datagram (see Large payloads under Transport). On the receiving side, a later `rmw_wait()` drains the bound socket, splits the header from the payload, and queues the message; `rmw_take` then deserializes it and hands it to the subscription callback. In steady state, the only discovery cost per publish is a single atomic read of the generation counter. @@ -299,7 +299,7 @@ Every entity in the system claims exactly one slot in the shared-memory registry A slot is the `RegistryEntrySlot` struct mapped into shared memory (`registry.hpp`). It is plain data with no pointers, because pointers would be meaningless across processes that map the region at different addresses. Two fields at the front coordinate concurrent access, and the rest describe the endpoint: - `seq` (32-bit atomic) — the seqlock counter. Even means the payload is stable; odd means a writer is mid-update. -- `state` (8-bit atomic) — the entity kind and lifecycle marker: `ENTRY_EMPTY`, `ENTRY_NODE`, `ENTRY_PUBLISHER`, `ENTRY_SUBSCRIPTION`, `ENTRY_SERVICE`, `ENTRY_CLIENT`, plus the internal `ENTRY_RESERVED` value a writer sets to claim an empty slot before its payload is filled in. +- `state` (8-bit atomic) — the entity kind and lifecycle marker: `ENTRY_EMPTY`, `ENTRY_NODE`, `ENTRY_PUBLISHER`, `ENTRY_SUBSCRIPTION`, `ENTRY_SERVICE`, `ENTRY_CLIENT`, `ENTRY_DOORBELL` (one per context, holding the socket path of that process's wakeup doorbell — see The doorbell under Wait set; type-filtered queries never match it, so it is invisible to graph introspection), plus the internal `ENTRY_RESERVED` value a writer sets to claim an empty slot before its payload is filled in. - `pid` — the owning process ID. This is the liveness handle. Stale-entry cleanup reclaims a slot when `/proc/` no longer exists. - `gid[16]` — the RMW GID, the 16-byte unique identity of this endpoint. - `node_name[256]` and `node_namespace[256]` — which node owns this endpoint, used to answer graph queries. @@ -392,7 +392,7 @@ slots[i].state.store(static_cast(entry.type), std::memory_order_release The release store pairs with the acquire load every reader does on `state`. Any reader that now observes a real type (`ENTRY_NODE`, `ENTRY_PUBLISHER`, and so on) is guaranteed to also see the fully-written payload that was published before it. This store is the single point at which the slot becomes visible to discovery. Until it happens, the slot reads as `ENTRY_RESERVED` and is invisible. -**Bump the generation counter.** Finally the writer does `header->generation.fetch_add(1)`. The generation counter is the table's change signal. Publishers, services, and clients cache their lookup results and re-scan only when the generation moves, so bumping it here tells every cached reader that the graph changed and its cache is stale. This is what turns a new registration into a graph event without any push notification or daemon. +**Bump the generation counter, then ring the doorbells.** Finally the writer does `header->generation.fetch_add(1)`. The generation counter is the table's change signal. Publishers, services, and clients cache their lookup results and re-scan only when the generation moves, so bumping it here tells every cached reader that the graph changed and its cache is stale. Immediately after the bump, the writer sends one octet to every registered doorbell (`ring_doorbells`), which wakes any process blocked in `rmw_wait` so it re-reads the counter. The order is load-bearing: ring strictly after the bump, paired with the wait side draining its doorbell strictly before reading the counter, is what makes a lost wakeup impossible (see The doorbell under Wait set). This is what turns a new registration into a graph event without any daemon: the table is the state, the doorbell is the edge. **The slot index is remembered, so removal needs no scan.** `registry_add` returns the slot index, and the owning entity stores it. Removal (`registry_remove`) indexes straight to that slot and CASes its `state` back to `ENTRY_EMPTY`. There is no second scan to find the entry on the way out, which keeps teardown cheap and makes removal symmetric with the single-CAS claim used on the way in. @@ -532,16 +532,35 @@ The ROS 2 executor finds out that work is ready by calling `rmw_wait()`. It hand Each `rmw_wait()` runs the same sequence on the calling thread: 1. **Drain first.** Every subscription, service, and client socket is drained into a per-entity message queue before anything blocks. The receive sockets are `SOCK_DGRAM | SOCK_NONBLOCK`, so the drain loop calls `recv_from` repeatedly and stops cleanly on `EAGAIN` (nothing left to read). This step exists because data may have arrived between the previous `rmw_wait()` and this one; draining up front means such data is not missed. -2. **Check the graph.** The shared-memory registry holds a `generation` counter that is bumped whenever an endpoint is added or removed. The wait reads it once and compares it against the value cached on the context. If it moved, the graph changed, and the context's graph guard condition is triggered. This is a single atomic load from shared memory, which is why a daemon or cross-process push notification is not needed (see the graph guard condition design choice). -3. **Arm the fds.** Every entity fd and guard-condition eventfd is added to the epoll instance with `EPOLL_CTL_ADD`. This is idempotent across calls: a still-live fd returns `EEXIST` and is treated as already armed, and a fd number reused after its previous owner closed gets freshly armed. The kernel removes closed fds from an epoll set automatically, so there is no matching `EPOLL_CTL_DEL` and no per-call teardown. -4. **Check ready, then block only if needed.** If any queue already holds a message, or any guard-condition eventfd reads as triggered, the call skips blocking entirely and reports what is ready. Only when nothing is ready does it call `epoll_wait()` with the computed timeout. The thread is asleep in the kernel here, not spinning. `epoll_wait()` is retried on `EINTR` against a steady-clock deadline so a signal neither returns a false timeout nor busy-loops. +2. **Check the graph.** The wait set carries its context (stored at `rmw_create_wait_set`), so this step runs for every wait — including a wait set that holds only guard conditions, which is exactly the shape rclcpp's GraphListener uses. The check first drains the context's doorbell socket, then reads the registry's `generation` counter and compares it against the value cached on the context. If it moved, the graph changed: any `TRANSIENT_LOCAL` publishers replay their cached messages to newly-matched subscribers, and every node's graph guard condition is triggered (see the next subsection). The drain-before-read order is half of the lost-wakeup proof; the other half is on the registry's writer side. +3. **Arm the fds.** Every entity fd, guard-condition eventfd, and the context's doorbell fd is added to the epoll instance with `EPOLL_CTL_ADD`. This is idempotent across calls: a still-live fd returns `EEXIST` and is treated as already armed, and a fd number reused after its previous owner closed gets freshly armed. The kernel removes closed fds from an epoll set automatically, so there is no matching `EPOLL_CTL_DEL` and no per-call teardown. +4. **Check ready, then block only if needed.** If any queue already holds a message, or any guard-condition eventfd reads as triggered, the call skips blocking entirely and reports what is ready. Only when nothing is ready does it block in `epoll_wait()`. The thread is asleep in the kernel here, not spinning; there is no internal poll interval. A wake caused only by the doorbell is internal: the wait re-runs the graph check of step 2 and goes back to sleep for the caller's remaining time. `RMW_RET_TIMEOUT` therefore surfaces only at the caller's own deadline, and an infinite wait blocks until something the caller asked about is actually ready. `epoll_wait()` is retried on `EINTR` against the same deadline so a signal neither returns a false timeout nor busy-loops. 5. **Drain again and report.** After waking, the sockets are drained once more, then the output arrays are pruned: entities with no pending data are set to `NULL`, and the ones that are ready are left in place for the executor to service. If nothing became ready, the call returns `RMW_RET_TIMEOUT`. ### Which guard condition the graph change wakes -There are two graph guard conditions in play, and the path between them matters. Each node creates its own guard condition in `rmw_create_node()` and stores it on the node (`UdsNode::graph_guard_condition`). `rmw_node_get_graph_guard_condition()` hands rcl that per-node object, so the per-node guard condition is what the executor's wait set actually watches for graph changes. +Each node creates its own graph guard condition in `rmw_create_node()` and stores it on the node (`UdsNode::graph_guard_condition`). `rmw_node_get_graph_guard_condition()` hands rcl that per-node object, so the per-node guard condition is what the executor's wait set actually watches for graph changes. -The trigger in step 2 above, however, fires `ctx->graph_guard_condition` (the context-level guard condition), not the per-node one. In the current code these are separate objects, and `UdsContext::graph_guard_condition` is never assigned, so the trigger does not reach the guard condition rcl is waiting on. A graph change is still observed, because every `rmw_wait()` re-reads the registry generation in step 2 and re-drains regardless of guard-condition state, so a wait already in progress or the next wait call picks up the change. The separate context-level trigger is therefore redundant rather than load-bearing. This is a wiring gap worth confirming against intent: if the context guard condition is meant to wake blocked waits on a graph change, it would need to be the node's guard condition (or be linked to it), and it would need to be assigned. +To reach it, the context keeps a mutex-guarded list of every node's graph guard condition. `rmw_create_node()` appends to the list as its last step (so no failure path has to undo it); `rmw_destroy_node()` removes the entry under the same mutex before destroying the guard condition, so the trigger loop can never fire a freed object. When step 2 of the wait observes a generation change, it triggers every guard condition in that list. A GraphListener blocked on a graph guard condition alone is woken by the doorbell (its fd is in the wait set's epoll), re-runs the check, triggers its own guard condition, and reports it ready — which is how `wait_for_service` and graph-change callbacks make progress with no data traffic at all. + +### The doorbell: cross-process wakeup without a daemon + +A single atomic load can tell a *running* wait that the graph changed, but it cannot wake a *blocked* one: an mmap store is invisible to `epoll`. Something a mutation can touch must be a file descriptor in the sleeping process. The doorbell is that object, chosen over the alternatives (signals, cross-process eventfd passing, inotify, io_uring futex — each fails on hygiene, permissions, coverage, or container seccomp) because it reuses the one primitive this transport is already made of: an `AF_UNIX` datagram socket. + +Each context binds one doorbell socket at `rmw_init` (a `ctl_*` file beside the data sockets) and registers it in the registry as `ENTRY_DOORBELL` — before taking its first generation snapshot, so no mutation can fall between the snapshot and the wiring. Every registry mutation (add, remove, stale-slot reclaim), after bumping the generation counter, sends one octet to every registered doorbell, non-blocking and best-effort. + +The correctness argument is one ordering pair. The writer rings strictly **after** the generation bump; the waiter drains its doorbell strictly **before** reading the generation. Any mutation therefore either lands in the generation value the waiter is about to read, or leaves a datagram queued on a level-triggered fd that makes the next `epoll_wait` return immediately. The datagram queues whether or not the target is currently blocked, so there is no check-then-block race to lose. The proof rests entirely on those two orderings; both call sites carry a comment saying so, and a regression test pins the behavior. + +Best-effort has sharp edges, each handled explicitly: + +- **A full receiver queue is success.** `EAGAIN` because the destination's queue is full means wakeups are already pending there; dropping the octet loses nothing. +- **A slow peer must not mute the others.** `AF_UNIX` datagrams stay charged to the *sender's* buffer until the receiver consumes them, so one process that never drains (hung, or not yet waiting) could exhaust the ring socket's budget and make sends fail for every peer. On `EAGAIN` the ring socket is closed, recreated, and the send retried once: on a fresh socket, `EAGAIN` can only mean the benign case above. A test that wedges one participant and asserts a healthy one still gets latched delivery guards this. +- **A recycled slot must not receive the octet.** The ring loop re-validates the slot type inside its seqlock window, so a slot that was a doorbell at the start of the scan but has been reused by a data endpoint cannot be sent to. +- **The ring socket is per mutating thread** (thread-local, closed at thread exit), so ringing takes no lock and leaks nothing. + +Cleanup needs no new machinery: the doorbell is a PID-owned slot holding a socket path, so graceful shutdown removes it like any endpoint, and the existing stale-PID reaper reclaims it after a crash — slot teardown already unlinks the socket file. + +Two limits are accepted. A build that predates the doorbell bumps the generation but never rings, so a fleet running mixed old and new builds can miss wakeups during the upgrade window — upgrade the fleet together, as with any wire-format change. And a process whose threads are never inside `rmw_wait` at all (an executor wedged in a user callback, or a bare-rmw publisher that never waits) cannot observe anything, doorbell or not; that gap predates the doorbell and is unchanged. ### In-process concurrency @@ -634,13 +653,11 @@ Request and response are correlated by sequence number, and that correlation is ### The graph guard condition -ROS 2 lets a node block until the graph changes (a node, publisher, subscription, service, or client appears or disappears). The mechanism is a graph guard condition: rcl adds it to a wait set, and the RMW arranges for that wait set to wake when the graph moves. This RMW detects graph changes by polling, not by cross-process push, because pushing a notification between processes without a daemon would need per-process signals or pipes, which is exactly the complexity the design avoids. - -The signal source is the registry's `generation` counter. Every successful add, remove, and stale-slot reclaim does `generation.fetch_add(1)` after publishing its slot change. A single monotonic counter in shared memory is enough to mean "something in the graph changed" without saying what. +ROS 2 lets a node block until the graph changes (a node, publisher, subscription, service, or client appears or disappears). The mechanism is a graph guard condition: rcl adds it to a wait set, and the RMW arranges for that wait set to wake when the graph moves. -Each node owns its own graph guard condition, created at `rmw_node_create` as an `eventfd` (see the wait section) and stored on the node. `rmw_node_get_graph_guard_condition` hands that per-node guard condition to rcl, so the object rcl waits on is the node's. On every `rmw_wait` the context reads the current `generation` and compares it to the value it cached on the previous call (`UdsContext::last_registry_generation`). When the counter has moved, the context records the new value, so the change is observed exactly once. Because rcl calls `rmw_wait` continuously while a node is spinning, a graph change is observed within one wait cycle, and the cost of the check is a single atomic load from shared memory. +The signal source is the registry's `generation` counter. Every successful add, remove, and stale-slot reclaim does `generation.fetch_add(1)` after publishing its slot change, then rings every registered doorbell (see The doorbell under Wait set). A single monotonic counter in shared memory is enough to mean "something in the graph changed" without saying what; the doorbell is what carries that fact into a process that is asleep. -There is a known seam here, the same one described under the wait set. The per-node guard condition is what rcl receives and waits on, but the trigger call in `rmw_wait` targets `UdsContext::graph_guard_condition`, a separate field that is never assigned and so is always null. The detection of the generation change is correct and the cached generation is advanced, but the explicit eventfd write that would wake a node blocked on the graph guard alone does not currently fire through that field. In practice the wait set is driven by its other fds and the per-cycle generation check, so graph queries observe the change; wiring the trigger to the per-node guard condition rcl actually holds is a correctness gap worth closing. +Each node owns its own graph guard condition, created at `rmw_node_create` as an `eventfd` (see the wait section), stored on the node, and registered in the context's list of graph guard conditions. `rmw_node_get_graph_guard_condition` hands that per-node guard condition to rcl, so the object rcl waits on is the node's — and it is exactly the object `rmw_wait` triggers when it observes a generation change (`UdsContext::last_registry_generation` records the new value, so each change is observed once per context). The doorbell wakes the blocked wait, the generation check runs, the per-node guard conditions fire, and rcl's graph machinery proceeds — with no polling interval anywhere in the path. ### GID generation @@ -668,6 +685,8 @@ Docker requirements: - `--pid=host` — for cross-container stale-PID cleanup correctness - `-v /tmp/ros2_uds:/tmp/ros2_uds` — for socket file sharing +Files under `/tmp/ros2_uds//` must not be removed while their owning processes are alive: they are live sockets (data and doorbell), not temp files, and an external cleaner deleting one silently severs delivery or wakeups to that process. Exempt the directory from tmpfile sweepers, e.g. a `tmpfiles.d` drop-in containing `x /tmp/ros2_uds`. (Ubuntu's default configuration cleans `/tmp` only at boot, and Docker containers run no cleaner, so this bites mainly on hosts with a cron-driven `tmpwatch`/`tmpreaper`.) + ## Resource usage, limitations, and build ### Resource profile @@ -702,6 +721,10 @@ This RMW communicates only between processes on a single host. `AF_UNIX` sockets Each message must fit in a single datagram; the usable cap is roughly 400 KB on a stock kernel. See The per-message size cap (~400 KB) under Transport for the full derivation and the sysctl remedy. +#### Notification requires a thread inside rmw_wait + +Graph events and `TRANSIENT_LOCAL` late-joiner replay are serviced from inside `rmw_wait` (woken by the doorbell). A process none of whose threads ever enters `rmw_wait` — an executor wedged in a user callback, or a bare-rmw publisher that never waits — cannot replay its latched messages or observe graph changes until it next waits or publishes. Standard rclcpp nodes always have a waiting thread (the GraphListener), so this bites only unusual bare-rmw setups. + #### Functions that return `RMW_RET_UNSUPPORTED` These functions are part of the RMW API but cannot be backed by a copy-based Unix-socket transport. Returning `RMW_RET_UNSUPPORTED` is the contract that tells rcl/rclcpp to skip the feature gracefully rather than fail. diff --git a/rmw_unix_socket_cpp/src/registry.cpp b/rmw_unix_socket_cpp/src/registry.cpp index b0187df..84080b3 100644 --- a/rmw_unix_socket_cpp/src/registry.cpp +++ b/rmw_unix_socket_cpp/src/registry.cpp @@ -21,7 +21,9 @@ #include #include +#include #include +#include #include #include "logging.hpp" @@ -29,6 +31,9 @@ namespace rmw_uds { +// Defined below; called after every generation bump (see its comment). +static void ring_doorbells(RegistryHeader * header); + // Lock-free atomics in shared memory require both lock-freedom AND // address-freedom. On every Linux target we support these hold; assert at // compile time so we fail loud on exotic platforms. @@ -276,6 +281,7 @@ static int32_t try_add_once(RegistryHeader * header, const RegistryEntry & entry !header->high_water_slot.compare_exchange_weak( cur, want, std::memory_order_relaxed, std::memory_order_relaxed)) {} header->generation.fetch_add(1, std::memory_order_acq_rel); + ring_doorbells(header); // strictly after the bump — see ring_doorbells return static_cast(i); } } @@ -345,6 +351,7 @@ void registry_remove(RegistryHeader * header, int32_t index) } teardown_slot(slot); header->generation.fetch_add(1, std::memory_order_acq_rel); + ring_doorbells(header); // strictly after the bump — see ring_doorbells } // Best-effort: stat /proc/. ENOENT means the PID is not in our @@ -372,13 +379,96 @@ static const char * entry_type_name(uint8_t t) case ENTRY_SUBSCRIPTION: return "subscription"; case ENTRY_SERVICE: return "service"; case ENTRY_CLIENT: return "client"; + case ENTRY_DOORBELL: return "doorbell"; default: return "?"; } } +// Ring every registered doorbell (one octet, best-effort) so processes blocked +// in rmw_wait re-check the registry. Called strictly AFTER a generation bump: +// paired with rmw_wait draining its doorbell strictly BEFORE reading the +// generation, every mutation either lands in the pre-block generation read or +// leaves a queued datagram on a level-triggered fd — no lost wakeup. EAGAIN +// means the peer already has a wakeup queued; other send errors mean a dead +// peer whose slot will be reclaimed. Scans slots directly (not registry_query, +// which calls back into cleanup and would recurse). +static void ring_doorbells(RegistryHeader * header) +{ + // One ring socket per mutating thread, closed at thread exit. AF_UNIX + // datagrams stay charged to the SENDER's buffer until the receiver consumes + // them, so a peer that is slow to drain could exhaust this fd's budget and + // make sendto fail for every OTHER peer too; the recreate-on-EAGAIN below + // resets that budget. On a fresh fd, EAGAIN can only mean the destination's + // own queue is full — a wakeup is already pending there, so the drop is safe. + struct RingFd + { + int fd = -1; + ~RingFd() {if (fd >= 0) {close(fd);}} + }; + static thread_local RingFd ring; + if (ring.fd < 0) { + ring.fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + if (ring.fd < 0) { + return; + } + } + auto * slots = registry_slots(header); + const uint32_t hi = header->high_water_slot.load(std::memory_order_acquire); + for (uint32_t i = 0; i < hi; ++i) { + if (slots[i].state.load(std::memory_order_acquire) != + static_cast(ENTRY_DOORBELL)) + { + continue; + } + // Seqlock snapshot of the socket path, re-validating the type INSIDE the + // seq window: without it, a remove + re-claim of this slot by a data + // endpoint between the fast-skip above and the copy could land the wake + // octet on a real data socket. A rewrite after this re-check still bumps + // seq, so the s1 comparison below rejects the torn copy. + char path[sizeof(slots[i].socket_path)]; + const uint32_t s1 = slots[i].seq.load(std::memory_order_acquire); + if (s1 & 1) { + continue; + } + if (slots[i].state.load(std::memory_order_acquire) != + static_cast(ENTRY_DOORBELL)) + { + continue; + } + std::memcpy(path, slots[i].socket_path, sizeof(path)); + if (slots[i].seq.load(std::memory_order_acquire) != s1 || path[0] == '\0') { + continue; + } + struct sockaddr_un addr; + std::memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + // memcpy of the measured length: addr is zeroed, so termination is free + // and -Wstringop-truncation stays quiet (path may fill all 108 bytes). + std::memcpy(addr.sun_path, path, strnlen(path, sizeof(addr.sun_path) - 1)); + const uint8_t octet = 1; + ssize_t sent = sendto( + ring.fd, &octet, 1, MSG_DONTWAIT | MSG_NOSIGNAL, + reinterpret_cast(&addr), sizeof(addr)); + if (sent < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + // Sender-side budget exhausted (some peer is slow to drain): reset the + // budget and retry once, so one undrained doorbell cannot mute rings to + // healthy peers. EAGAIN again on the fresh fd is the benign case. + close(ring.fd); + ring.fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + if (ring.fd < 0) { + return; + } + (void)sendto( + ring.fd, &octet, 1, MSG_DONTWAIT | MSG_NOSIGNAL, + reinterpret_cast(&addr), sizeof(addr)); + } + } +} + void registry_cleanup_stale(RegistryHeader * header) { auto * slots = registry_slots(header); + bool reclaimed = false; // Scan only [0, high_water): over-scan is safe, under-scan is impossible. uint32_t hw = header->high_water_slot.load(std::memory_order_acquire); uint32_t max = header->max_entries; @@ -435,6 +525,10 @@ void registry_cleanup_stale(RegistryHeader * header) teardown_slot(&slots[i]); header->generation.fetch_add(1, std::memory_order_acq_rel); + reclaimed = true; + } + if (reclaimed) { + ring_doorbells(header); // strictly after the bump(s) — see ring_doorbells } } diff --git a/rmw_unix_socket_cpp/src/registry.hpp b/rmw_unix_socket_cpp/src/registry.hpp index b4d9141..25c4b9d 100644 --- a/rmw_unix_socket_cpp/src/registry.hpp +++ b/rmw_unix_socket_cpp/src/registry.hpp @@ -44,6 +44,11 @@ enum RegistryEntryType : uint8_t ENTRY_SUBSCRIPTION, ENTRY_SERVICE, ENTRY_CLIENT, + // Per-context wakeup socket (see ring_doorbells in registry.cpp): rung with + // one octet after every registry mutation so a blocked rmw_wait re-checks + // the registry. Additive: no slot layout change, and type-filtered queries + // never match it, so it is invisible to graph introspection. + ENTRY_DOORBELL, // Transient claim state: a writer won the slot but has not yet committed its // payload. Readers treat it like ENTRY_EMPTY so they never observe a slot // before its payload is published. Never stored in a RegistryEntry; lives diff --git a/rmw_unix_socket_cpp/src/rmw_init.cpp b/rmw_unix_socket_cpp/src/rmw_init.cpp index 9d89780..9a3a281 100644 --- a/rmw_unix_socket_cpp/src/rmw_init.cpp +++ b/rmw_unix_socket_cpp/src/rmw_init.cpp @@ -184,8 +184,44 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context) "rmw_init: context up (domain_id=%zu, pid=%d)", domain_id, static_cast(getpid())); - // Read initial generation auto * header = rmw_uds::registry_header(ctx->registry_ptr); + + // Doorbell: bind + register BEFORE the first generation snapshot below, so + // no registry mutation can land in the gap between the snapshot and the + // wakeup wiring (a mutation after registration rings this socket; one + // before it is covered by the snapshot). + { + const std::string ctl_path = rmw_uds::make_socket_path(domain_id, "ctl"); + ctx->doorbell_fd = rmw_uds::create_bound_socket(ctl_path); + if (ctx->doorbell_fd < 0) { + RMW_UDS_LOG_ERROR( + "rmw_init: failed to create doorbell socket (domain_id=%zu)", domain_id); + close(ctx->send_socket_fd); + rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size); + delete ctx; + RMW_SET_ERROR_MSG("failed to create doorbell socket"); + return RMW_RET_ERROR; + } + rmw_uds::RegistryEntry dentry; + std::memset(&dentry, 0, sizeof(dentry)); + dentry.type = rmw_uds::ENTRY_DOORBELL; + dentry.pid = getpid(); + std::strncpy(dentry.socket_path, ctl_path.c_str(), sizeof(dentry.socket_path) - 1); + ctx->doorbell_registry_index = rmw_uds::registry_add(header, dentry); + if (ctx->doorbell_registry_index < 0) { + RMW_UDS_LOG_ERROR( + "rmw_init: registry full — cannot register doorbell (domain_id=%zu)", domain_id); + close(ctx->doorbell_fd); + unlink(ctl_path.c_str()); + close(ctx->send_socket_fd); + rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size); + delete ctx; + RMW_SET_ERROR_MSG("registry full — cannot register doorbell"); + return RMW_RET_ERROR; + } + } + + // Read initial generation ctx->last_registry_generation.store( rmw_uds::registry_generation(header), std::memory_order_relaxed); @@ -195,6 +231,8 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context) if (options->enclave) { enclave_copy = rcutils_strdup(options->enclave, options->allocator); if (!enclave_copy) { + rmw_uds::registry_remove(header, ctx->doorbell_registry_index); + close(ctx->doorbell_fd); close(ctx->send_socket_fd); rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size); delete ctx; @@ -220,6 +258,8 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context) if (enclave_copy) { options->allocator.deallocate(enclave_copy, options->allocator.state); } + rmw_uds::registry_remove(header, ctx->doorbell_registry_index); + close(ctx->doorbell_fd); close(ctx->send_socket_fd); rmw_uds::registry_close(ctx->registry_fd, ctx->registry_ptr, ctx->registry_size); delete ctx; @@ -255,6 +295,15 @@ rmw_ret_t rmw_context_fini(rmw_context_t * context) auto * ctx = reinterpret_cast(context->impl); if (ctx) { + // Doorbell teardown before the registry unmaps: registry_remove's slot + // teardown also unlinks the socket file. + if (ctx->doorbell_registry_index >= 0 && ctx->registry_ptr) { + auto * header = rmw_uds::registry_header(ctx->registry_ptr); + rmw_uds::registry_remove(header, ctx->doorbell_registry_index); + } + if (ctx->doorbell_fd >= 0) { + close(ctx->doorbell_fd); + } if (ctx->send_socket_fd >= 0) { close(ctx->send_socket_fd); } diff --git a/rmw_unix_socket_cpp/src/rmw_node.cpp b/rmw_unix_socket_cpp/src/rmw_node.cpp index a788cdc..06f53f3 100644 --- a/rmw_unix_socket_cpp/src/rmw_node.cpp +++ b/rmw_unix_socket_cpp/src/rmw_node.cpp @@ -16,7 +16,9 @@ #include "registry.hpp" #include "types.hpp" +#include #include +#include #include "rcutils/strdup.h" #include "rmw/allocators.h" @@ -108,6 +110,13 @@ rmw_node_t * rmw_create_node( return nullptr; } + // Last step, so no failure path above needs to undo it: expose the graph GC + // to rmw_wait's generation check (triggered there on graph changes). + { + std::lock_guard lock(ctx->graph_gcs_mutex); + ctx->graph_gcs.push_back(graph_gc); + } + return node; } @@ -127,6 +136,15 @@ rmw_ret_t rmw_destroy_node(rmw_node_t * node) } if (node_data->graph_guard_condition) { + // Unpublish from the context BEFORE destroying, so rmw_wait can never + // trigger a freed guard condition (it holds the same mutex). + if (node_data->context) { + std::lock_guard lock(node_data->context->graph_gcs_mutex); + auto & gcs = node_data->context->graph_gcs; + gcs.erase( + std::remove(gcs.begin(), gcs.end(), node_data->graph_guard_condition), + gcs.end()); + } auto _r [[maybe_unused]] = rmw_destroy_guard_condition(node_data->graph_guard_condition); } diff --git a/rmw_unix_socket_cpp/src/rmw_wait.cpp b/rmw_unix_socket_cpp/src/rmw_wait.cpp index 5941c35..2d80fdb 100644 --- a/rmw_unix_socket_cpp/src/rmw_wait.cpp +++ b/rmw_unix_socket_cpp/src/rmw_wait.cpp @@ -17,6 +17,7 @@ #include "transport.hpp" #include "types.hpp" +#include #include #include #include @@ -27,6 +28,7 @@ #include #include +#include #include #include "rmw/allocators.h" @@ -97,6 +99,7 @@ rmw_wait_set_t * rmw_create_wait_set(rmw_context_t * context, size_t max_conditi RMW_SET_ERROR_MSG("failed to allocate wait set data"); return nullptr; } + ws_data->context = reinterpret_cast(context->impl); ws_data->epoll_fd = epoll_create1(EPOLL_CLOEXEC); if (ws_data->epoll_fd < 0) { @@ -182,18 +185,26 @@ rmw_ret_t rmw_wait( } } - // 2. Check graph generation changes — trigger graph guard conditions - // We need to find the context from any available entity - rmw_uds::UdsContext * ctx = nullptr; - if (subscriptions && subscriptions->subscriber_count > 0 && subscriptions->subscribers[0]) { - ctx = static_cast(subscriptions->subscribers[0])->context; - } else if (services && services->service_count > 0 && services->services[0]) { - ctx = static_cast(services->services[0])->context; - } else if (clients && clients->client_count > 0 && clients->clients[0]) { - ctx = static_cast(clients->clients[0])->context; - } - - if (ctx && ctx->registry_ptr) { + // 2. Check graph generation changes — trigger graph guard conditions. + // The context comes from the wait set itself (set at rmw_create_wait_set): + // a guard-condition-only wait set (rclcpp's GraphListener) has no entity to + // scavenge it from, and this check must run for those waits too. Wrapped in + // a lambda so the step-4 loop can re-run it on each doorbell wake. + rmw_uds::UdsContext * ctx = ws_data->context; + const int doorbell_fd = ctx ? ctx->doorbell_fd : -1; + auto run_generation_check = [&]() { + // Drain the doorbell strictly BEFORE reading the generation: paired with + // ring_doorbells running strictly AFTER the bump, a mutation either lands + // in this generation read or leaves a queued datagram that keeps the + // level-triggered fd readable — no lost wakeup. + if (doorbell_fd >= 0) { + uint8_t buf[16]; + while (recv(doorbell_fd, buf, sizeof(buf), MSG_DONTWAIT) > 0) { + } + } + if (!(ctx && ctx->registry_ptr)) { + return; + } auto * header = rmw_uds::registry_header(ctx->registry_ptr); uint64_t gen = rmw_uds::registry_generation(header); if (gen != ctx->last_registry_generation.load(std::memory_order_relaxed)) { @@ -256,20 +267,19 @@ rmw_ret_t rmw_wait( } } - // Trigger all graph guard conditions in the guard_conditions list - if (guard_conditions) { - for (size_t i = 0; i < guard_conditions->guard_condition_count; ++i) { - if (!guard_conditions->guard_conditions[i]) {continue;} - // We don't know which are graph GCs, so we just note the change - // The graph GC is triggered by the node itself + // Wake graph listeners: trigger every node's graph guard condition + // (rclcpp's GraphListener waits on these). rmw_destroy_node removes a + // node's GC from this list under the same mutex before destroying it, + // so a freed guard condition is never triggered. + { + std::lock_guard gc_lock(ctx->graph_gcs_mutex); + for (auto * gc : ctx->graph_gcs) { + auto _r [[maybe_unused]] = rmw_trigger_guard_condition(gc); } } - // Trigger graph guard condition on the context - if (ctx->graph_guard_condition) { - auto _r [[maybe_unused]] = rmw_trigger_guard_condition(ctx->graph_guard_condition); - } } - } + }; + run_generation_check(); // Arm every entity fd with epoll on every wait. EPOLL_CTL_ADD is idempotent // here: a still-live fd returns EEXIST (already armed), while a fd number @@ -319,6 +329,8 @@ rmw_ret_t rmw_wait( register_fd(gc->eventfd_fd); } } + // The context's doorbell: rung by any process after a registry mutation. + register_fd(doorbell_fd); } // 3. Check if anything is already ready @@ -395,39 +407,57 @@ rmw_ret_t rmw_wait( } } - // Bound the wait so an idle executor loops and re-runs the top-of-wait - // TRANSIENT_LOCAL late-joiner replay: a joining subscriber only bumps the - // shm generation counter (no fd fires), so an unbounded wait would never - // replay to it. Only shortens the timeout; a non-blocking 0 stays 0. - constexpr int TL_REPLAY_POLL_MS = 200; - if (timeout_ms < 0 || timeout_ms > TL_REPLAY_POLL_MS) { - timeout_ms = TL_REPLAY_POLL_MS; - } - - // Block, retrying on EINTR. A finite timeout uses a steady_clock deadline - // so a signal interruption neither returns TIMEOUT early nor busy-loops. + // Block until something the caller waits on fires, or the caller's own + // deadline. There is no internal poll: a registry mutation in any process + // rings this context's doorbell (ring_doorbells in registry.cpp), which + // wakes the epoll; the doorbell is drained, the registry re-checked + // (TRANSIENT_LOCAL late-joiner replay + graph guard conditions), and — if + // nothing the caller waits on became ready — the wait re-blocks. + // RMW_RET_TIMEOUT surfaces only at the caller's own deadline; an infinite + // wait never surfaces a synthetic timeout. EINTR re-enters the loop, so a + // signal neither returns TIMEOUT early nor busy-loops. + const bool infinite = (timeout_ms < 0); + const int64_t caller_deadline_ns = + infinite ? 0 : now_ns() + static_cast(timeout_ms) * 1000000; struct epoll_event ready_events[64]; - const int64_t deadline_ns = - (timeout_ms >= 0) ? now_ns() + static_cast(timeout_ms) * 1000000 : 0; - int remaining_ms = timeout_ms; while (true) { - int n = epoll_wait(ws_data->epoll_fd, ready_events, 64, remaining_ms); - if (n >= 0) { - break; + int block_ms = -1; + if (!infinite) { + const int64_t rem_ns = caller_deadline_ns - now_ns(); + const int64_t rem_ms = (rem_ns > 0) ? (rem_ns + 999999) / 1000000 : 0; // ceil + block_ms = static_cast( + std::min(rem_ms, std::numeric_limits::max())); } - if (errno != EINTR) { + int n = epoll_wait(ws_data->epoll_fd, ready_events, 64, block_ms); + if (n < 0) { + if (errno == EINTR) { + continue; + } RMW_SET_ERROR_MSG("epoll_wait failed"); return RMW_RET_ERROR; } - if (timeout_ms < 0) { - continue; // Infinite wait: just re-block. + if (n == 0) { + break; // The caller's deadline passed -> timeout; fall through to drain. + } + bool only_doorbell = true; + bool rang = false; + for (int e = 0; e < n; ++e) { + if (ready_events[e].data.fd == doorbell_fd) { + rang = true; + } else { + only_doorbell = false; + } + } + if (rang) { + run_generation_check(); // Drains the doorbell, replays, triggers GCs. + } + if (!only_doorbell) { + break; // Something the caller waits on fired -> fall through to drain. } - const int64_t rem_ns = deadline_ns - now_ns(); - if (rem_ns <= 0) { - break; // Deadline passed -> timeout; fall through to drain. + if (!infinite && now_ns() >= caller_deadline_ns) { + break; // Doorbell-only wake at the deadline -> timeout. } - const int64_t rem_ms = rem_ns / 1000000; - remaining_ms = (rem_ms > 0) ? static_cast(rem_ms) : 1; // >=1ms while time remains + // Doorbell-only wake: re-block for the caller's remaining time. } // No EPOLL_CTL_DEL needed — fds stay registered across calls. diff --git a/rmw_unix_socket_cpp/src/transport.cpp b/rmw_unix_socket_cpp/src/transport.cpp index 9842a1b..17aed6f 100644 --- a/rmw_unix_socket_cpp/src/transport.cpp +++ b/rmw_unix_socket_cpp/src/transport.cpp @@ -105,6 +105,16 @@ int create_send_socket() SEND_BUF_SIZE, std::strerror(errno)); } + // Test seam: override the send buffer so tests can pin kernel size-cap + // behavior (EMSGSIZE) deterministically. Never set in production. + const char * test_sndbuf = std::getenv("RMW_UDS_TEST_SNDBUF"); + if (test_sndbuf != nullptr) { + int forced = std::atoi(test_sndbuf); + if (forced > 0) { + (void)setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &forced, sizeof(forced)); + } + } + return fd; } diff --git a/rmw_unix_socket_cpp/src/types.hpp b/rmw_unix_socket_cpp/src/types.hpp index fa40803..dc3091b 100644 --- a/rmw_unix_socket_cpp/src/types.hpp +++ b/rmw_unix_socket_cpp/src/types.hpp @@ -119,7 +119,19 @@ struct UdsContext int send_socket_fd = -1; std::atomic is_shutdown{false}; std::atomic last_registry_generation{0}; - rmw_guard_condition_t * graph_guard_condition = nullptr; + + // Doorbell: a bound datagram socket other processes ring (one octet) after + // any registry mutation, so a blocked rmw_wait re-checks the registry + // without polling. Registered in the registry as ENTRY_DOORBELL; the slot's + // teardown unlinks the socket file (graceful or via the stale-PID reaper). + int doorbell_fd = -1; + int32_t doorbell_registry_index = -1; + + // Per-node graph guard conditions (see rmw_node_get_graph_guard_condition), + // triggered from rmw_wait when the registry generation changes. Guarded by + // the mutex; rmw_destroy_node removes its entry before destroying the GC. + std::mutex graph_gcs_mutex; + std::vector graph_gcs; // TRANSIENT_LOCAL publishers, for wait-side cache replay on graph change. std::mutex transient_local_pubs_mutex; @@ -301,6 +313,10 @@ struct UdsGuardCondition struct UdsWaitSet { int epoll_fd = -1; + // Set at rmw_create_wait_set. The top-of-wait replay/graph check needs the + // context even when the wait set holds only guard conditions (rclcpp's + // GraphListener), so it cannot be scavenged from the waited-on entities. + UdsContext * context = nullptr; }; } // namespace rmw_uds diff --git a/rmw_unix_socket_cpp/test/test_internal_publisher_cache.cpp b/rmw_unix_socket_cpp/test/test_internal_publisher_cache.cpp new file mode 100644 index 0000000..e7c01d5 --- /dev/null +++ b/rmw_unix_socket_cpp/test/test_internal_publisher_cache.cpp @@ -0,0 +1,150 @@ +// Copyright 2026 Abderahmane BENALI +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Deliberate white-box tests of the TL replay cache's pruning: the oracle is +// the publisher's private known_subscriber_paths set (a leak there has no +// public-API observable short of unbounded memory growth), so these inspect +// UdsPublisher internals on purpose. Behavioral QoS coverage lives in +// test_rmw_qos.cpp. + +#include "test_base.hpp" + +#include + +#include "test_msgs/msg/basic_types.hpp" + +#include "rosidl_typesupport_cpp/message_type_support.hpp" + +#include "../src/types.hpp" + +class QosTest : public RmwUdsNodeTest +{ +protected: + const rosidl_message_type_support_t * ts = nullptr; + + void SetUp() override + { + RmwUdsNodeTest::SetUp(); + ts = rosidl_typesupport_cpp::get_message_type_support_handle< + test_msgs::msg::BasicTypes>(); + } + + rmw_qos_profile_t make_qos( + rmw_qos_reliability_policy_e rel, + rmw_qos_durability_policy_e dur, + size_t depth = 10) + { + rmw_qos_profile_t qos; + std::memset(&qos, 0, sizeof(qos)); + qos.history = RMW_QOS_POLICY_HISTORY_KEEP_LAST; + qos.depth = depth; + qos.reliability = rel; + qos.durability = dur; + return qos; + } +}; + +TEST_F(QosTest, KnownSubscriberPathsPrunedOnChurn) +{ + // The publisher's known_subscriber_paths must not accumulate dead entries as + // subscribers churn: each create/destroy bumps the registry generation, and a + // restarted subscriber gets a brand-new unique socket path. Without pruning on + // refresh the set is insert-only and grows by one per churned subscriber. + auto qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, + RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher(node, ts, "/churn", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + // Seed the cache so there is something to replay. + test_msgs::msg::BasicTypes seed; + seed.int32_value = 1; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &seed, nullptr)); + + auto sub_opts = rmw_get_default_subscription_options(); + constexpr int kChurn = 8; + for (int i = 0; i < kChurn; ++i) { + // New sub bumps generation -> next publish refreshes + records this sub. + auto * sub = rmw_create_subscription(node, ts, "/churn", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + test_msgs::msg::BasicTypes m; + m.int32_value = i + 2; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + // Destroy bumps generation again; next publish refreshes + prunes the gone sub. + auto _r [[maybe_unused]] = rmw_destroy_subscription(node, sub); + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + } + + // After the loop every churned sub is destroyed, so known should be empty. + auto * pub_data = static_cast(pub->data); + size_t known_size = 0; + { + std::lock_guard lock(pub_data->cache_mutex); + known_size = pub_data->known_subscriber_paths.size(); + } + // With the prune: tracks only live subs (0 here). Without it: grows to kChurn. + EXPECT_LE(known_size, 1u) + << "known_subscriber_paths leaked dead entries: size=" << known_size; + + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); +} + +TEST_F(QosTest, TransientLocalSerializedKnownSubscriberPathsPrunedOnChurn) +{ + // Same prune guarantee as KnownSubscriberPathsPrunedOnChurn, but driven + // through rmw_publish_serialized_message, which carries its own copy of the + // prune-on-refresh logic. Guards against that copy silently diverging: without + // the prune the serialized path's known_subscriber_paths grows by one per + // churned subscriber. + auto qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, + RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher(node, ts, "/churn_serialized", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + uint8_t bytes[] = {1, 2, 3, 4, 5, 6, 7, 8}; + rmw_serialized_message_t msg; + msg.buffer = bytes; + msg.buffer_length = sizeof(bytes); + msg.buffer_capacity = sizeof(bytes); + msg.allocator = rcutils_get_default_allocator(); + + // Seed the cache so there is something to replay. + EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &msg, nullptr)); + + auto sub_opts = rmw_get_default_subscription_options(); + constexpr int kChurn = 8; + for (int i = 0; i < kChurn; ++i) { + auto * sub = rmw_create_subscription(node, ts, "/churn_serialized", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &msg, nullptr)); + auto _r [[maybe_unused]] = rmw_destroy_subscription(node, sub); + EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &msg, nullptr)); + } + + auto * pub_data = static_cast(pub->data); + size_t known_size = 0; + { + std::lock_guard lock(pub_data->cache_mutex); + known_size = pub_data->known_subscriber_paths.size(); + } + EXPECT_LE(known_size, 1u) + << "serialized-path known_subscriber_paths leaked dead entries: size=" << known_size; + + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); +} diff --git a/rmw_unix_socket_cpp/test/test_registry.cpp b/rmw_unix_socket_cpp/test/test_registry.cpp index 4fd7788..169c4b1 100644 --- a/rmw_unix_socket_cpp/test/test_registry.cpp +++ b/rmw_unix_socket_cpp/test/test_registry.cpp @@ -218,10 +218,10 @@ TEST_F(RegistryTest, MultipleEntriesSameType) } // The high-water bound must shrink the scan range without dropping the -// highest-index entry. Adding sequentially from an empty (fresh SetUp) -// registry fills slots 0..4, so high_water must equal max(idx)+1 = the count. -// Fails on unmodified main (no high_water_slot member) and on a buggy fix -// that scans [0, hw) with hw==idx or [0, hw-1). +// highest-index entry. The invariant is hw >= max(live idx)+1 so a [0, hw) +// scan covers every live slot; exact equality is a first-fit detail we +// deliberately don't pin. Fails on unmodified main (no high_water_slot +// member) and on a buggy fix that scans [0, hw) with hw==idx or [0, hw-1). TEST_F(RegistryTest, HighWaterBoundsScanWithoutLosingTopEntry) { auto * header = rmw_uds::registry_header(registry_ptr); @@ -237,9 +237,9 @@ TEST_F(RegistryTest, HighWaterBoundsScanWithoutLosingTopEntry) idx.push_back(k); } - // high_water is the count (slots fill 0..4), i.e. max(idx)+1. + // hw must cover the highest live slot: scanning [0, hw) cannot lose it. uint32_t hw = header->high_water_slot.load(); - EXPECT_EQ(static_cast(*std::max_element(idx.begin(), idx.end())) + 1, hw); + EXPECT_GE(hw, static_cast(*std::max_element(idx.begin(), idx.end())) + 1); EXPECT_LT(hw, header->max_entries); // bound far below 32768 -> scan is cheap // The top (highest-index) entry must still be found -> guards the [0, hw) vs @@ -407,6 +407,13 @@ TEST_F(RegistryTest, AddOverflowKeepsLivePidEntries) auto all = rmw_uds::registry_query( header, rmw_uds::ENTRY_NODE, nullptr, nullptr, nullptr); + // The live entries actually survived: exactly n_0, n_2 and extra remain. + EXPECT_EQ(3u, all.size()); + for (const char * name : {"n_0", "n_2", "extra"}) { + auto found = rmw_uds::registry_query( + header, rmw_uds::ENTRY_NODE, nullptr, name, nullptr); + EXPECT_EQ(1u, found.size()) << "live entry '" << name << "' lost"; + } // All remaining entries must have a live PID (ours) — no dead PIDs left. for (const auto & r : all) { EXPECT_NE(0, std::strncmp(r.node_name.c_str(), "n_1", 3)) diff --git a/rmw_unix_socket_cpp/test/test_registry_concurrent.cpp b/rmw_unix_socket_cpp/test/test_registry_concurrent.cpp index c488f66..6eb09b2 100644 --- a/rmw_unix_socket_cpp/test/test_registry_concurrent.cpp +++ b/rmw_unix_socket_cpp/test/test_registry_concurrent.cpp @@ -187,14 +187,27 @@ TEST_F(RegistryConcurrentTest, ReadersNeverSeeTornEntries) readers.emplace_back(reader); } - std::this_thread::sleep_for(std::chrono::milliseconds(1500)); + // Run until a meaningful sample is collected, with a hard cap so a + // starved CI runner stalls the loop instead of failing the floor assert. + const int observation_floor = 1000; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + bool capped = false; + while (total_observations.load(std::memory_order_relaxed) < observation_floor) { + if (std::chrono::steady_clock::now() >= deadline) { + capped = true; + break; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } stop.store(true, std::memory_order_relaxed); w.join(); for (auto & r : readers) { r.join(); } - EXPECT_GT(total_observations.load(), 0) << "test did not exercise the path"; + if (!capped) { + EXPECT_GE(total_observations.load(), observation_floor); + } EXPECT_EQ(0, torn_observations.load()) << "observed " << torn_observations.load() << " torn snapshots out of " << total_observations.load(); @@ -302,6 +315,10 @@ TEST_F(RegistryConcurrentTest, GenerationCounterMonotonicUnderChaos) auto * header = rmw_uds::registry_header(registry_ptr); std::atomic stop{false}; std::atomic regressions{0}; + std::atomic mutations{0}; + std::atomic reads{0}; + + uint64_t gen_before = rmw_uds::registry_generation(header); auto mutator = [&](int tid) { int n = 0; @@ -314,6 +331,7 @@ TEST_F(RegistryConcurrentTest, GenerationCounterMonotonicUnderChaos) int32_t idx = rmw_uds::registry_add(header, e); if (idx >= 0) { rmw_uds::registry_remove(header, idx); + mutations.fetch_add(2, std::memory_order_relaxed); // add + remove } } }; @@ -326,6 +344,7 @@ TEST_F(RegistryConcurrentTest, GenerationCounterMonotonicUnderChaos) regressions.fetch_add(1, std::memory_order_relaxed); } local_last = g; + reads.fetch_add(1, std::memory_order_relaxed); } }; @@ -333,10 +352,24 @@ TEST_F(RegistryConcurrentTest, GenerationCounterMonotonicUnderChaos) for (int i = 0; i < 8; ++i) {threads.emplace_back(mutator, i);} for (int i = 0; i < 4; ++i) {threads.emplace_back(observer);} - std::this_thread::sleep_for(std::chrono::milliseconds(300)); + // Run until real work has accumulated, with a hard cap for starved runners. + const uint64_t progress_floor = 1000; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while ((mutations.load(std::memory_order_relaxed) < progress_floor || + reads.load(std::memory_order_relaxed) < progress_floor) && + std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } stop.store(true, std::memory_order_relaxed); for (auto & t : threads) {t.join();} EXPECT_EQ(0, regressions.load()) << "generation counter regressed " << regressions.load() << " times"; + // Real progress, not a vacuous pass: mutations happened, observers watched, + // and every successful mutation bumped the counter. + uint64_t gen_after = rmw_uds::registry_generation(header); + EXPECT_GT(mutations.load(), 0u) << "no mutations performed"; + EXPECT_GT(reads.load(), 0u) << "observer never read the counter"; + EXPECT_GE(gen_after - gen_before, mutations.load()); } diff --git a/rmw_unix_socket_cpp/test/test_rmw_graph.cpp b/rmw_unix_socket_cpp/test/test_rmw_graph.cpp index edd4eac..87c3a98 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_graph.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_graph.cpp @@ -122,19 +122,38 @@ TEST_F(RmwUdsNodeTest, GetTopicNamesAndTypes) EXPECT_EQ(RMW_RET_OK, rmw_destroy_publisher(node, pub)); } -TEST_F(RmwUdsNodeTest, CompareGidsEqual) +TEST_F(RmwUdsNodeTest, CompareGidsRealPublishers) { + auto * ts = rosidl_typesupport_cpp::get_message_type_support_handle< + test_msgs::msg::BasicTypes>(); + rmw_qos_profile_t qos; + std::memset(&qos, 0, sizeof(qos)); + qos.history = RMW_QOS_POLICY_HISTORY_KEEP_LAST; + qos.depth = 10; + qos.reliability = RMW_QOS_POLICY_RELIABILITY_RELIABLE; + qos.durability = RMW_QOS_POLICY_DURABILITY_VOLATILE; + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub1 = rmw_create_publisher(node, ts, "/gid_topic", &qos, &pub_opts); + ASSERT_NE(nullptr, pub1); + auto * pub2 = rmw_create_publisher(node, ts, "/gid_topic", &qos, &pub_opts); + ASSERT_NE(nullptr, pub2); + rmw_gid_t gid1, gid2; - std::memset(&gid1, 0, sizeof(gid1)); - std::memset(&gid2, 0, sizeof(gid2)); - gid1.data[0] = 1; - gid2.data[0] = 1; + ASSERT_EQ(RMW_RET_OK, rmw_get_gid_for_publisher(pub1, &gid1)); + ASSERT_EQ(RMW_RET_OK, rmw_get_gid_for_publisher(pub2, &gid2)); + // Each gid equals itself bool result = false; - EXPECT_EQ(RMW_RET_OK, rmw_compare_gids_equal(&gid1, &gid2, &result)); + EXPECT_EQ(RMW_RET_OK, rmw_compare_gids_equal(&gid1, &gid1, &result)); + EXPECT_TRUE(result); + EXPECT_EQ(RMW_RET_OK, rmw_compare_gids_equal(&gid2, &gid2, &result)); EXPECT_TRUE(result); - gid2.data[0] = 2; + // Distinct publishers have distinct gids EXPECT_EQ(RMW_RET_OK, rmw_compare_gids_equal(&gid1, &gid2, &result)); EXPECT_FALSE(result); + + EXPECT_EQ(RMW_RET_OK, rmw_destroy_publisher(node, pub1)); + EXPECT_EQ(RMW_RET_OK, rmw_destroy_publisher(node, pub2)); } diff --git a/rmw_unix_socket_cpp/test/test_rmw_guard_condition.cpp b/rmw_unix_socket_cpp/test/test_rmw_guard_condition.cpp index 8b0c26d..62c2b38 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_guard_condition.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_guard_condition.cpp @@ -14,11 +14,6 @@ #include "test_base.hpp" -#include -#include - -#include "../src/types.hpp" - TEST_F(RmwUdsTestBase, CreateDestroyGuardCondition) { auto * gc = rmw_create_guard_condition(&context); @@ -26,9 +21,6 @@ TEST_F(RmwUdsTestBase, CreateDestroyGuardCondition) EXPECT_EQ(uds_id(), gc->implementation_identifier); EXPECT_NE(nullptr, gc->data); - auto * data = static_cast(gc->data); - EXPECT_GE(data->eventfd_fd, 0); - EXPECT_EQ(RMW_RET_OK, rmw_destroy_guard_condition(gc)); } @@ -37,15 +29,35 @@ TEST_F(RmwUdsTestBase, TriggerGuardCondition) auto * gc = rmw_create_guard_condition(&context); ASSERT_NE(nullptr, gc); + auto * ws = rmw_create_wait_set(&context, 1); + ASSERT_NE(nullptr, ws); + EXPECT_EQ(RMW_RET_OK, rmw_trigger_guard_condition(gc)); - // Verify the eventfd is readable - auto * data = static_cast(gc->data); - uint64_t val = 0; - ssize_t r = read(data->eventfd_fd, &val, sizeof(val)); - EXPECT_EQ(static_cast(sizeof(val)), r); - EXPECT_GT(val, 0u); + // A wait after the trigger must return immediately with the entry ready + rmw_guard_conditions_t guard_conditions; + void * gc_array[1] = {gc->data}; + guard_conditions.guard_conditions = gc_array; + guard_conditions.guard_condition_count = 1; + + rmw_time_t timeout; + timeout.sec = 0; + timeout.nsec = 100000000; // 100ms + + rmw_ret_t ret = rmw_wait(nullptr, &guard_conditions, nullptr, nullptr, nullptr, ws, &timeout); + EXPECT_EQ(RMW_RET_OK, ret); + EXPECT_NE(nullptr, guard_conditions.guard_conditions[0]); + + // The trigger is one-shot: a second wait must time out with the entry nulled + gc_array[0] = gc->data; + timeout.sec = 0; + timeout.nsec = 50000000; // 50ms + + ret = rmw_wait(nullptr, &guard_conditions, nullptr, nullptr, nullptr, ws, &timeout); + EXPECT_EQ(RMW_RET_TIMEOUT, ret); + EXPECT_EQ(nullptr, guard_conditions.guard_conditions[0]); + EXPECT_EQ(RMW_RET_OK, rmw_destroy_wait_set(ws)); EXPECT_EQ(RMW_RET_OK, rmw_destroy_guard_condition(gc)); } diff --git a/rmw_unix_socket_cpp/test/test_rmw_pub_sub.cpp b/rmw_unix_socket_cpp/test/test_rmw_pub_sub.cpp index bb01ff1..8e47780 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_pub_sub.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_pub_sub.cpp @@ -17,6 +17,7 @@ #include #include "test_msgs/msg/basic_types.hpp" +#include "test_msgs/msg/empty.hpp" #include "test_msgs/msg/strings.hpp" #include "test_msgs/msg/unbounded_sequences.hpp" @@ -25,13 +26,6 @@ #include #include -#include - -#include -#include -#include - -#include "../src/types.hpp" // UdsSubscription + WireHeader layouts only (no linked symbols) class PubSubTest : public RmwUdsNodeTest { @@ -204,7 +198,7 @@ TEST_F(PubSubTest, PublisherGetGid) EXPECT_FALSE(all_zero); } -TEST_F(PubSubTest, LargeMessageViaShmRing) +TEST_F(PubSubTest, LargeMessageRoundtrip) { // A payload well above SHM_PAYLOAD_THRESHOLD travels through the // publisher's /dev/shm ring: the datagram carries only a descriptor and @@ -227,10 +221,6 @@ TEST_F(PubSubTest, LargeMessageViaShmRing) } EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &send_msg, nullptr)); - // The publisher must actually have created its ring segment. - auto * pub_data = static_cast(pub->data); - EXPECT_NE(nullptr, pub_data->shm_ring.base); - test_msgs::msg::UnboundedSequences recv_msg; bool taken = false; EXPECT_EQ(RMW_RET_OK, rmw_take(sub, &recv_msg, &taken, nullptr)); @@ -262,56 +252,41 @@ TEST_F(PubSubTest, StringMessages) // Regression for the rmw_take_sequence cursor bug: a deserialize failure in the // middle of a batch must not leave an uninitialized hole or miscount size — each -// success is written at the *taken cursor, not the loop index. We inject -// [good, corrupt, good] straight into the subscription's datagram socket via raw -// POSIX sendto (a single SOCK_DGRAM sender preserves order, so the corrupt one -// lands between the two good ones). Before the fix the second good message was -// written at data[2] and lost, while size == *taken == 2 exposed data[1] (the -// hole from the failed deserialize) to the consumer as a valid message. +// success is written at the *taken cursor, not the loop index. We produce +// [good, corrupt, good] entirely through the public API: the corrupt element +// comes from a second publisher on the same topic with a smaller message type +// (Empty), whose CDR payload is too short to deserialize as BasicTypes. Both +// publishers share the node's send socket, so a single SOCK_DGRAM sender +// preserves order and the mismatched datagram lands between the two good ones. +// Before the fix the second good message was written at data[2] and lost, while +// size == *taken == 2 exposed data[1] (the hole from the failed deserialize) to +// the consumer as a valid message. TEST_F(PubSubTest, TakeSequenceSkipsMidBatchCorruptContiguously) { auto sub_opts = rmw_get_default_subscription_options(); sub = rmw_create_subscription(node, ts, "/take_seq_corrupt", &qos, &sub_opts); ASSERT_NE(nullptr, sub); - auto * sub_data = static_cast(sub->data); - ASSERT_FALSE(sub_data->socket_path.empty()); - // A valid wire payload via the public rmw_serialize API (identical CDR bytes - // to what a publisher emits, so the take path deserializes it cleanly). + auto pub_opts = rmw_get_default_publisher_options(); + pub = rmw_create_publisher(node, ts, "/take_seq_corrupt", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + // Same topic, smaller type: fan-out matches by topic only, so its datagram + // reaches the subscriber but fails CDR deserialization as BasicTypes. + auto empty_ts = rosidl_typesupport_cpp::get_message_type_support_handle< + test_msgs::msg::Empty>(); + rmw_publisher_t * corrupt_pub = + rmw_create_publisher(node, empty_ts, "/take_seq_corrupt", &qos, &pub_opts); + ASSERT_NE(nullptr, corrupt_pub); + test_msgs::msg::BasicTypes good; good.int32_value = 4242; good.bool_value = true; - rcutils_allocator_t allocator = rcutils_get_default_allocator(); - rmw_serialized_message_t good_ser = rmw_get_zero_initialized_serialized_message(); - ASSERT_EQ(RMW_RET_OK, rmw_serialized_message_init(&good_ser, 0, &allocator)); - ASSERT_EQ(RMW_RET_OK, rmw_serialize(&good, ts, &good_ser)); - - int send_fd = ::socket(AF_UNIX, SOCK_DGRAM, 0); - ASSERT_GE(send_fd, 0); - struct sockaddr_un addr; - std::memset(&addr, 0, sizeof(addr)); - addr.sun_family = AF_UNIX; - std::strncpy(addr.sun_path, sub_data->socket_path.c_str(), sizeof(addr.sun_path) - 1); - - // Each datagram is the packed WireHeader followed by the payload (recv_from's framing). - auto inject = [&](const uint8_t * payload, size_t len) { - rmw_uds::WireHeader hdr; - std::memset(&hdr, 0, sizeof(hdr)); - hdr.payload_size = static_cast(len); - hdr.msg_type = 0; // topic message - std::vector dgram(sizeof(hdr) + len); - std::memcpy(dgram.data(), &hdr, sizeof(hdr)); - if (len > 0) {std::memcpy(dgram.data() + sizeof(hdr), payload, len);} - ASSERT_EQ( - static_cast(dgram.size()), - ::sendto(send_fd, dgram.data(), dgram.size(), 0, - reinterpret_cast(&addr), sizeof(addr))); - }; - - const uint8_t corrupt[4] = {0xDE, 0xAD, 0xBE, 0xEF}; // too short to deserialize - inject(good_ser.buffer, good_ser.buffer_length); - inject(corrupt, sizeof(corrupt)); - inject(good_ser.buffer, good_ser.buffer_length); + test_msgs::msg::Empty mismatched; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &good, nullptr)); + EXPECT_EQ(RMW_RET_OK, rmw_publish(corrupt_pub, &mismatched, nullptr)); + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &good, nullptr)); + EXPECT_EQ(RMW_RET_OK, rmw_destroy_publisher(node, corrupt_pub)); std::this_thread::sleep_for(std::chrono::milliseconds(50)); // defensive; datagrams already buffered @@ -339,7 +314,4 @@ TEST_F(PubSubTest, TakeSequenceSkipsMidBatchCorruptContiguously) EXPECT_EQ(2u, info_seq.size); EXPECT_EQ(4242, out[0].int32_value); EXPECT_EQ(4242, out[1].int32_value); // the bug left this a hole and lost it - - ::close(send_fd); - EXPECT_EQ(RMW_RET_OK, rmw_serialized_message_fini(&good_ser)); } diff --git a/rmw_unix_socket_cpp/test/test_rmw_qos.cpp b/rmw_unix_socket_cpp/test/test_rmw_qos.cpp index 02571d0..2487f67 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_qos.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_qos.cpp @@ -20,7 +20,7 @@ #include #include -#include +#include #include #include "test_msgs/msg/basic_types.hpp" @@ -31,7 +31,7 @@ #include "rosidl_typesupport_cpp/message_type_support.hpp" #include "rosidl_typesupport_cpp/service_type_support.hpp" -#include "types.hpp" +#include "../src/shm_transport.hpp" // SHM_PAYLOAD_THRESHOLD only (compile-time size guards) class QosTest : public RmwUdsNodeTest { @@ -60,6 +60,15 @@ class QosTest : public RmwUdsNodeTest } }; +// Sets an env var for the enclosing scope; unset even on an early ASSERT return. +struct ScopedEnv +{ + const char * name; + ScopedEnv(const char * n, const char * v) + : name(n) {setenv(n, v, 1);} + ~ScopedEnv() {unsetenv(name);} +}; + // --- TRANSIENT_LOCAL (latched) tests --- TEST_F(QosTest, TransientLocalLateJoiner) @@ -109,17 +118,11 @@ TEST_F(QosTest, TransientLocalLateJoiner) auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); } -TEST_F(QosTest, TransientLocalLargeMessageUsesDurableShm) +TEST_F(QosTest, TransientLocalLargeMessageReachesLateJoiner) { - // Large latched messages are staged into a dedicated *durable* shm segment - // owned by the cache entry — never the publisher's cycling ring, which would - // lap and corrupt a record still awaited by a late joiner. So the cached - // entry carries a descriptor (SHM_PAYLOAD_FLAG set), shm_ring stays untouched, - // and replay resolves the descriptor out of the durable segment. This also - // pins the ordering in rmw_publish: the TL branch must stage durably rather - // than fall through to the ring fork. 100 KB: over SHM_PAYLOAD_THRESHOLD, - // under the stock-kernel datagram cap. (See TransientLocalHugeMessageLateJoiner - // for the case that exceeds the cap.) + // A large latched message — 100 KB: over SHM_PAYLOAD_THRESHOLD, under the + // stock-kernel datagram cap — must reach a late joiner byte-equal. (See + // TransientLocalHugeMessageLateJoiner for the case that exceeds the cap.) auto seq_ts = rosidl_typesupport_cpp::get_message_type_support_handle< test_msgs::msg::UnboundedSequences>(); auto qos = make_qos( @@ -137,19 +140,7 @@ TEST_F(QosTest, TransientLocalLargeMessageUsesDurableShm) } EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &msg, nullptr)); - auto * pub_data = static_cast(pub->data); - { - std::lock_guard lock(pub_data->cache_mutex); - ASSERT_EQ(1u, pub_data->message_cache.size()); - const auto & cm = pub_data->message_cache.back(); - EXPECT_NE(nullptr, cm.shm_seg) - << "a large TRANSIENT_LOCAL message must be cached in a durable shm segment"; - EXPECT_TRUE(cm.header.msg_type & rmw_uds::SHM_PAYLOAD_FLAG); - } - EXPECT_EQ(nullptr, pub_data->shm_ring.base) - << "TRANSIENT_LOCAL must use a durable segment, never the cycling ring"; - - // A late joiner must get the cached message via durable-shm replay. + // A late joiner must get the cached message. auto sub_opts = rmw_get_default_subscription_options(); auto * sub = rmw_create_subscription(node, seq_ts, "/latched_large", &qos, &sub_opts); ASSERT_NE(nullptr, sub); @@ -157,7 +148,6 @@ TEST_F(QosTest, TransientLocalLargeMessageUsesDurableShm) test_msgs::msg::UnboundedSequences trigger; trigger.uint8_values = {1, 2, 3}; EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &trigger, nullptr)); - EXPECT_EQ(nullptr, pub_data->shm_ring.base); test_msgs::msg::UnboundedSequences recv; bool taken = false; @@ -199,13 +189,6 @@ TEST_F(QosTest, TransientLocalHugeMessageLateJoiner) } EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &big, nullptr)); - auto * pub_data = static_cast(pub->data); - { - std::lock_guard lock(pub_data->cache_mutex); - ASSERT_EQ(1u, pub_data->message_cache.size()); - EXPECT_NE(nullptr, pub_data->message_cache.back().shm_seg); - } - // Late joiner, then a small publish to trigger replay of the cached 5 MB msg. auto sub_opts = rmw_get_default_subscription_options(); auto * sub = rmw_create_subscription(node, seq_ts, "/latched_huge", &qos, &sub_opts); @@ -232,17 +215,18 @@ TEST_F(QosTest, TransientLocalHugeMessageLateJoiner) TEST_F(QosTest, TransientLocalPublishReturnsErrorOnEMSGSIZE) { // A latched publish whose live send is rejected by the kernel size cap must - // return RMW_RET_ERROR, not a lying RMW_RET_OK. Before the fix the TL path - // ignored send_to's result entirely. Shrink SO_SNDBUF so a sub-threshold - // (inline) latched message hits EMSGSIZE deterministically, independent of - // the machine's net.core.wmem_max. - auto * ctx_impl = reinterpret_cast(context.impl); - int small_buf = 2048; - ASSERT_EQ( - 0, - setsockopt( - ctx_impl->send_socket_fd, SOL_SOCKET, SO_SNDBUF, - &small_buf, sizeof(small_buf))); + // return RMW_RET_ERROR, not a lying RMW_RET_OK. The RMW_UDS_TEST_SNDBUF init + // seam shrinks SO_SNDBUF so a sub-threshold (inline) latched message hits + // EMSGSIZE deterministically, independent of the machine's net.core.wmem_max. + // The seam applies at socket creation, so the test brings up its own context. + ScopedEnv sndbuf("RMW_UDS_TEST_SNDBUF", "2048"); + rmw_init_options_t opts2 = rmw_get_zero_initialized_init_options(); + ASSERT_EQ(RMW_RET_OK, rmw_init_options_init(&opts2, rcutils_get_default_allocator())); + opts2.domain_id = 99; // same domain as the fixture + rmw_context_t ctx2 = rmw_get_zero_initialized_context(); + ASSERT_EQ(RMW_RET_OK, rmw_init(&opts2, &ctx2)); + rmw_node_t * node2 = rmw_create_node(&ctx2, "tl_emsgsize_node", "/test_ns"); + ASSERT_NE(nullptr, node2); auto seq_ts = rosidl_typesupport_cpp::get_message_type_support_handle< test_msgs::msg::UnboundedSequences>(); @@ -250,10 +234,10 @@ TEST_F(QosTest, TransientLocalPublishReturnsErrorOnEMSGSIZE) RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); auto pub_opts = rmw_get_default_publisher_options(); - auto * pub = rmw_create_publisher(node, seq_ts, "/tl_emsgsize", &qos, &pub_opts); + auto * pub = rmw_create_publisher(node2, seq_ts, "/tl_emsgsize", &qos, &pub_opts); ASSERT_NE(nullptr, pub); auto sub_opts = rmw_get_default_subscription_options(); - auto * sub = rmw_create_subscription(node, seq_ts, "/tl_emsgsize", &qos, &sub_opts); + auto * sub = rmw_create_subscription(node2, seq_ts, "/tl_emsgsize", &qos, &sub_opts); ASSERT_NE(nullptr, sub); // 32 KiB: above the shrunken buffer, below SHM_PAYLOAD_THRESHOLD so it stays @@ -263,14 +247,17 @@ TEST_F(QosTest, TransientLocalPublishReturnsErrorOnEMSGSIZE) msg.uint8_values.resize(32 * 1024); EXPECT_EQ(RMW_RET_ERROR, rmw_publish(pub, &msg, nullptr)); - auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); - auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node2, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node2, pub); + EXPECT_EQ(RMW_RET_OK, rmw_destroy_node(node2)); + EXPECT_EQ(RMW_RET_OK, rmw_shutdown(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_context_fini(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_init_options_fini(&opts2)); } -TEST_F(QosTest, TransientLocalLargeMessageInlineFallbackWhenShmUnavailable) +TEST_F(QosTest, TransientLocalLargeMessageDeliveredWhenShmUnavailable) { - // When shm staging is unavailable, a large latched payload falls back to - // caching inline (no durable segment, no SHM_PAYLOAD_FLAG) and is still + // When shm staging is unavailable, a large latched payload must still be // delivered byte-equal to a late joiner. The test seam forces the failure. auto seq_ts = rosidl_typesupport_cpp::get_message_type_support_handle< test_msgs::msg::UnboundedSequences>(); @@ -290,16 +277,6 @@ TEST_F(QosTest, TransientLocalLargeMessageInlineFallbackWhenShmUnavailable) EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &msg, nullptr)); unsetenv("RMW_UDS_TEST_FORCE_SHM_FAILURE"); // reset before it leaks to other tests - auto * pub_data = static_cast(pub->data); - { - std::lock_guard lock(pub_data->cache_mutex); - ASSERT_EQ(1u, pub_data->message_cache.size()); - const auto & cm = pub_data->message_cache.back(); - EXPECT_EQ(nullptr, cm.shm_seg) - << "shm forced unavailable — the large payload must be cached inline"; - EXPECT_FALSE(cm.header.msg_type & rmw_uds::SHM_PAYLOAD_FLAG); - } - auto sub_opts = rmw_get_default_subscription_options(); auto * sub = rmw_create_subscription(node, seq_ts, "/tl_fallback", &qos, &sub_opts); ASSERT_NE(nullptr, sub); @@ -310,20 +287,18 @@ TEST_F(QosTest, TransientLocalLargeMessageInlineFallbackWhenShmUnavailable) test_msgs::msg::UnboundedSequences recv; bool taken = false; EXPECT_EQ(RMW_RET_OK, rmw_take(sub, &recv, &taken, nullptr)); - ASSERT_TRUE(taken) << "late joiner must receive the inline-fallback message"; + ASSERT_TRUE(taken) << "late joiner must receive the message despite shm being unavailable"; EXPECT_EQ(msg.uint8_values, recv.uint8_values); auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); } -TEST_F(QosTest, VolatileLargeMessageInlineFallbackWhenShmUnavailable) +TEST_F(QosTest, VolatileLargeMessageDeliveredWhenShmUnavailable) { - // Ring-path counterpart of the TL test above: when shm staging is - // unavailable, a large VOLATILE payload is serialized into the inline - // fallback (no ring, no SHM_PAYLOAD_FLAG) and still delivered byte-equal. - // Pins shm_serialize_prepare_send's reserve-failure branch, including the - // contained resize on the extern "C" boundary. + // VOLATILE counterpart of the TL test above: when shm staging is + // unavailable, a large volatile payload must still be delivered byte-equal. + // The test seam forces the failure. auto seq_ts = rosidl_typesupport_cpp::get_message_type_support_handle< test_msgs::msg::UnboundedSequences>(); auto qos = make_qos( @@ -345,14 +320,10 @@ TEST_F(QosTest, VolatileLargeMessageInlineFallbackWhenShmUnavailable) EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &msg, nullptr)); unsetenv("RMW_UDS_TEST_FORCE_SHM_FAILURE"); // reset before it leaks to other tests - auto * pub_data = static_cast(pub->data); - EXPECT_EQ(nullptr, pub_data->shm_ring.base) - << "shm forced unavailable — no ring may be created"; - test_msgs::msg::UnboundedSequences recv; bool taken = false; EXPECT_EQ(RMW_RET_OK, rmw_take(sub, &recv, &taken, nullptr)); - ASSERT_TRUE(taken) << "the inline-fallback message must be delivered"; + ASSERT_TRUE(taken) << "the message must be delivered despite shm being unavailable"; EXPECT_EQ(msg.uint8_values, recv.uint8_values); auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); @@ -383,14 +354,8 @@ TEST_F(QosTest, TransientLocalSerializedLargeMessageLateJoiner) } serialized.allocator = rcutils_get_default_allocator(); - // Publish BEFORE any subscriber — must be cached in a durable segment. + // Publish BEFORE any subscriber — must be cached for replay. EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &serialized, nullptr)); - auto * pub_data = static_cast(pub->data); - { - std::lock_guard lock(pub_data->cache_mutex); - ASSERT_EQ(1u, pub_data->message_cache.size()); - EXPECT_NE(nullptr, pub_data->message_cache.back().shm_seg); - } auto sub_opts = rmw_get_default_subscription_options(); auto * sub = rmw_create_subscription(node, ts, "/tl_serialized_huge", &qos, &sub_opts); @@ -422,100 +387,6 @@ TEST_F(QosTest, TransientLocalSerializedLargeMessageLateJoiner) auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); } -TEST_F(QosTest, KnownSubscriberPathsPrunedOnChurn) -{ - // The publisher's known_subscriber_paths must not accumulate dead entries as - // subscribers churn: each create/destroy bumps the registry generation, and a - // restarted subscriber gets a brand-new unique socket path. Without pruning on - // refresh the set is insert-only and grows by one per churned subscriber. - auto qos = make_qos( - RMW_QOS_POLICY_RELIABILITY_RELIABLE, - RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); - - auto pub_opts = rmw_get_default_publisher_options(); - auto * pub = rmw_create_publisher(node, ts, "/churn", &qos, &pub_opts); - ASSERT_NE(nullptr, pub); - - // Seed the cache so there is something to replay. - test_msgs::msg::BasicTypes seed; - seed.int32_value = 1; - EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &seed, nullptr)); - - auto sub_opts = rmw_get_default_subscription_options(); - constexpr int kChurn = 8; - for (int i = 0; i < kChurn; ++i) { - // New sub bumps generation -> next publish refreshes + records this sub. - auto * sub = rmw_create_subscription(node, ts, "/churn", &qos, &sub_opts); - ASSERT_NE(nullptr, sub); - test_msgs::msg::BasicTypes m; - m.int32_value = i + 2; - EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); - // Destroy bumps generation again; next publish refreshes + prunes the gone sub. - auto _r [[maybe_unused]] = rmw_destroy_subscription(node, sub); - EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); - } - - // After the loop every churned sub is destroyed, so known should be empty. - auto * pub_data = static_cast(pub->data); - size_t known_size = 0; - { - std::lock_guard lock(pub_data->cache_mutex); - known_size = pub_data->known_subscriber_paths.size(); - } - // With the prune: tracks only live subs (0 here). Without it: grows to kChurn. - EXPECT_LE(known_size, 1u) - << "known_subscriber_paths leaked dead entries: size=" << known_size; - - auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); -} - -TEST_F(QosTest, TransientLocalSerializedKnownSubscriberPathsPrunedOnChurn) -{ - // Same prune guarantee as KnownSubscriberPathsPrunedOnChurn, but driven - // through rmw_publish_serialized_message, which carries its own copy of the - // prune-on-refresh logic. Guards against that copy silently diverging: without - // the prune the serialized path's known_subscriber_paths grows by one per - // churned subscriber. - auto qos = make_qos( - RMW_QOS_POLICY_RELIABILITY_RELIABLE, - RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); - - auto pub_opts = rmw_get_default_publisher_options(); - auto * pub = rmw_create_publisher(node, ts, "/churn_serialized", &qos, &pub_opts); - ASSERT_NE(nullptr, pub); - - uint8_t bytes[] = {1, 2, 3, 4, 5, 6, 7, 8}; - rmw_serialized_message_t msg; - msg.buffer = bytes; - msg.buffer_length = sizeof(bytes); - msg.buffer_capacity = sizeof(bytes); - msg.allocator = rcutils_get_default_allocator(); - - // Seed the cache so there is something to replay. - EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &msg, nullptr)); - - auto sub_opts = rmw_get_default_subscription_options(); - constexpr int kChurn = 8; - for (int i = 0; i < kChurn; ++i) { - auto * sub = rmw_create_subscription(node, ts, "/churn_serialized", &qos, &sub_opts); - ASSERT_NE(nullptr, sub); - EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &msg, nullptr)); - auto _r [[maybe_unused]] = rmw_destroy_subscription(node, sub); - EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &msg, nullptr)); - } - - auto * pub_data = static_cast(pub->data); - size_t known_size = 0; - { - std::lock_guard lock(pub_data->cache_mutex); - known_size = pub_data->known_subscriber_paths.size(); - } - EXPECT_LE(known_size, 1u) - << "serialized-path known_subscriber_paths leaked dead entries: size=" << known_size; - - auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); -} - TEST_F(QosTest, TransientLocalCacheDepthEnforced) { auto qos = make_qos( @@ -554,10 +425,11 @@ TEST_F(QosTest, TransientLocalCacheDepthEnforced) received.push_back(recv.int32_value); } - // We should have received 4, 5, 6 (the 3 most recent in cache + current) - ASSERT_GE(received.size(), 3u); - // The last received should be 6 (current message) - EXPECT_EQ(6, received.back()); + // Exactly 4, 5, 6 — the depth-3 cache tail plus the current message. + ASSERT_EQ(3u, received.size()); + EXPECT_EQ(4, received[0]); + EXPECT_EQ(5, received[1]); + EXPECT_EQ(6, received[2]); auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); @@ -567,25 +439,27 @@ TEST_F(QosTest, TransientLocalCacheDepthEnforced) TEST_F(QosTest, PublishReturnsErrorOnEMSGSIZE) { - // Shrink SO_SNDBUF so any send hits EMSGSIZE — publish must return ERROR. - auto * ctx_impl = reinterpret_cast(context.impl); - int small_buf = 2048; - ASSERT_EQ( - 0, - setsockopt( - ctx_impl->send_socket_fd, SOL_SOCKET, SO_SNDBUF, - &small_buf, sizeof(small_buf))); + // Shrink SO_SNDBUF (RMW_UDS_TEST_SNDBUF init seam, own context) so any send + // hits EMSGSIZE — publish must return ERROR. + ScopedEnv sndbuf("RMW_UDS_TEST_SNDBUF", "2048"); + rmw_init_options_t opts2 = rmw_get_zero_initialized_init_options(); + ASSERT_EQ(RMW_RET_OK, rmw_init_options_init(&opts2, rcutils_get_default_allocator())); + opts2.domain_id = 99; // same domain as the fixture + rmw_context_t ctx2 = rmw_get_zero_initialized_context(); + ASSERT_EQ(RMW_RET_OK, rmw_init(&opts2, &ctx2)); + rmw_node_t * node2 = rmw_create_node(&ctx2, "emsgsize_node", "/test_ns"); + ASSERT_NE(nullptr, node2); auto qos = make_qos( RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_VOLATILE); auto pub_opts = rmw_get_default_publisher_options(); - auto * pub = rmw_create_publisher(node, ts, "/emsgsize", &qos, &pub_opts); + auto * pub = rmw_create_publisher(node2, ts, "/emsgsize", &qos, &pub_opts); ASSERT_NE(nullptr, pub); auto sub_opts = rmw_get_default_subscription_options(); - auto * sub = rmw_create_subscription(node, ts, "/emsgsize", &qos, &sub_opts); + auto * sub = rmw_create_subscription(node2, ts, "/emsgsize", &qos, &sub_opts); ASSERT_NE(nullptr, sub); // 32 KB — well above SOCK_MIN_SNDBUF the kernel will clamp us to, but @@ -605,34 +479,38 @@ TEST_F(QosTest, PublishReturnsErrorOnEMSGSIZE) EXPECT_EQ(RMW_RET_ERROR, rmw_publish_serialized_message(pub, &serialized, nullptr)); std::free(serialized.buffer); - auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); - auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node2, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node2, pub); + EXPECT_EQ(RMW_RET_OK, rmw_destroy_node(node2)); + EXPECT_EQ(RMW_RET_OK, rmw_shutdown(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_context_fini(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_init_options_fini(&opts2)); } TEST_F(QosTest, LargePayloadBypassesSendBuffer) { - // Same shrunken SO_SNDBUF, but a payload above SHM_PAYLOAD_THRESHOLD: - // the bytes travel through the publisher's shm ring and only a small - // descriptor crosses the socket, so the publish succeeds and the message - // arrives intact where it previously died with EMSGSIZE. - auto * ctx_impl = reinterpret_cast(context.impl); - int small_buf = 2048; - ASSERT_EQ( - 0, - setsockopt( - ctx_impl->send_socket_fd, SOL_SOCKET, SO_SNDBUF, - &small_buf, sizeof(small_buf))); + // Same shrunken SO_SNDBUF (RMW_UDS_TEST_SNDBUF init seam, own context), but + // a payload above SHM_PAYLOAD_THRESHOLD: the publish succeeds and the + // message arrives intact where the inline path died with EMSGSIZE. + ScopedEnv sndbuf("RMW_UDS_TEST_SNDBUF", "2048"); + rmw_init_options_t opts2 = rmw_get_zero_initialized_init_options(); + ASSERT_EQ(RMW_RET_OK, rmw_init_options_init(&opts2, rcutils_get_default_allocator())); + opts2.domain_id = 99; // same domain as the fixture + rmw_context_t ctx2 = rmw_get_zero_initialized_context(); + ASSERT_EQ(RMW_RET_OK, rmw_init(&opts2, &ctx2)); + rmw_node_t * node2 = rmw_create_node(&ctx2, "shm_bypass_node", "/test_ns"); + ASSERT_NE(nullptr, node2); auto qos = make_qos( RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_VOLATILE); auto pub_opts = rmw_get_default_publisher_options(); - auto * pub = rmw_create_publisher(node, ts, "/shm_bypass", &qos, &pub_opts); + auto * pub = rmw_create_publisher(node2, ts, "/shm_bypass", &qos, &pub_opts); ASSERT_NE(nullptr, pub); auto sub_opts = rmw_get_default_subscription_options(); - auto * sub = rmw_create_subscription(node, ts, "/shm_bypass", &qos, &sub_opts); + auto * sub = rmw_create_subscription(node2, ts, "/shm_bypass", &qos, &sub_opts); ASSERT_NE(nullptr, sub); constexpr size_t big_size = 128 * 1024; @@ -661,8 +539,12 @@ TEST_F(QosTest, LargePayloadBypassesSendBuffer) auto _f [[maybe_unused]] = rmw_serialized_message_fini(&received); std::free(serialized.buffer); - auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); - auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node2, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node2, pub); + EXPECT_EQ(RMW_RET_OK, rmw_destroy_node(node2)); + EXPECT_EQ(RMW_RET_OK, rmw_shutdown(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_context_fini(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_init_options_fini(&opts2)); } TEST_F(QosTest, TransientLocalReplayOnWaitNoSubsequentPublish) @@ -723,7 +605,8 @@ TEST_F(QosTest, TransientLocalReplayOnWaitNoSubsequentPublish) TEST_F(QosTest, PublishStillReturnsOkOnSoftDropPeerGone) { - // ENOENT on a vanished peer must stay RET_OK — only EMSGSIZE escalates. + // A subscriber process that dies without cleanup must stay a soft drop: + // publishing to the dead peer returns RET_OK — only EMSGSIZE escalates. auto qos = make_qos( RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_VOLATILE); @@ -732,22 +615,47 @@ TEST_F(QosTest, PublishStillReturnsOkOnSoftDropPeerGone) auto * pub = rmw_create_publisher(node, ts, "/peer_gone", &qos, &pub_opts); ASSERT_NE(nullptr, pub); - auto sub_opts = rmw_get_default_subscription_options(); - auto * sub = rmw_create_subscription(node, ts, "/peer_gone", &qos, &sub_opts); - ASSERT_NE(nullptr, sub); + // Real dead peer: a forked child subscribes on its own context, signals over + // the pipe, then _exit(0)s without cleanup (skipping atexit/destructors). + int ready[2]; + ASSERT_EQ(0, pipe(ready)); + pid_t pid = fork(); + ASSERT_GE(pid, 0); + if (pid == 0) { + // Child: no gtest asserts; report failure through the exit code. + close(ready[0]); + rmw_init_options_t c_opts = rmw_get_zero_initialized_init_options(); + if (rmw_init_options_init(&c_opts, rcutils_get_default_allocator()) != RMW_RET_OK) { + _exit(1); + } + c_opts.domain_id = 99; // same domain as the fixture + rmw_context_t c_ctx = rmw_get_zero_initialized_context(); + if (rmw_init(&c_opts, &c_ctx) != RMW_RET_OK) {_exit(1);} + rmw_node_t * c_node = rmw_create_node(&c_ctx, "peer_gone_child", "/test_ns"); + if (c_node == nullptr) {_exit(1);} + auto c_sub_opts = rmw_get_default_subscription_options(); + auto * c_sub = rmw_create_subscription(c_node, ts, "/peer_gone", &qos, &c_sub_opts); + if (c_sub == nullptr) {_exit(1);} + char b = 'x'; + if (write(ready[1], &b, 1) != 1) {_exit(1);} + _exit(0); // dead peer: no destroy, no shutdown, no unregister + } - // Warm pub's path cache, then unlink the sub's socket → next sendmsg = ENOENT. + // Parent: wait until the child's subscription exists, then reap the corpse. + close(ready[1]); + char b = 0; + ASSERT_EQ(1, read(ready[0], &b, 1)) << "child died before subscribing"; + close(ready[0]); + int status = 0; + ASSERT_EQ(pid, waitpid(pid, &status, 0)); + ASSERT_TRUE(WIFEXITED(status)); + ASSERT_EQ(0, WEXITSTATUS(status)); + + // Publish to the vanished subscriber — must stay RET_OK. test_msgs::msg::BasicTypes m; - m.int32_value = 1; - ASSERT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); - - auto * sub_impl = static_cast(sub->data); - unlink(sub_impl->socket_path.c_str()); - m.int32_value = 2; EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); - auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); } @@ -980,10 +888,8 @@ TEST_F(QosTest, MultipleClientsOneService) TEST_F(QosTest, TransientLocalReplayReachesLateJoinerWhileWaitBlocked) { // The subscriber joins AFTER the publisher's executor is already blocked in - // rmw_wait. A joining subscriber only bumps the shm generation counter, which - // signals no fd, so an idle rmw_wait(infinite) would block in epoll forever - // and never re-run the top-of-wait replay. The latched message must still - // reach the late joiner within a bounded time. + // an unbounded rmw_wait. The latched message must still reach the late + // joiner within a bounded time. auto qos = make_qos( RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); @@ -1065,3 +971,73 @@ TEST_F(QosTest, TransientLocalReplayReachesLateJoinerWhileWaitBlocked) auto _r4 [[maybe_unused]] = rmw_destroy_guard_condition(gc); auto _r5 [[maybe_unused]] = rmw_destroy_service(node, srv); } + +TEST_F(QosTest, TransientLocalLateJoinerWhilePublisherProcessIdle) +{ + // Scenario: a latched (TRANSIENT_LOCAL) publisher lives in a process that + // is completely idle — its only executor thread is parked in an unbounded + // rmw_wait that contains no subscriptions, services, or clients. A + // subscriber that joins later must still receive the retained message. + // This is the user-visible bug: a latched topic on a quiet node never + // reaching late subscribers, however the process happens to be waiting. + auto qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, + RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); + + auto * gc = rmw_create_guard_condition(&context); + ASSERT_NE(nullptr, gc); + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher(node, ts, "/idle_replay_gc_only", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + test_msgs::msg::BasicTypes m; + m.int32_value = 9; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + + auto * ws = rmw_create_wait_set(&context, 4); + ASSERT_NE(nullptr, ws); + + std::atomic stop{false}; + std::thread executor( + [&] { + while (!stop.load()) { + void * gc_array[1] = {gc->data}; + rmw_guard_conditions_t gcs; + gcs.guard_conditions = gc_array; + gcs.guard_condition_count = 1; + auto _r [[maybe_unused]] = rmw_wait( + nullptr, &gcs, nullptr, nullptr, nullptr, ws, nullptr); + } + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(150)); + + auto sub_opts = rmw_get_default_subscription_options(); + auto * sub = rmw_create_subscription(node, ts, "/idle_replay_gc_only", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + + bool got = false; + for (int i = 0; i < 300 && !got; ++i) { + test_msgs::msg::BasicTypes recv; + bool taken = false; + if (rmw_take(sub, &recv, &taken, nullptr) == RMW_RET_OK && taken && + recv.int32_value == 9) + { + got = true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + stop.store(true); + auto _t [[maybe_unused]] = rmw_trigger_guard_condition(gc); + executor.join(); + + EXPECT_TRUE(got) << "late joiner never received the latched message while the " + "publisher's process was idle"; + + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_wait_set(ws); + auto _r3 [[maybe_unused]] = rmw_destroy_publisher(node, pub); + auto _r4 [[maybe_unused]] = rmw_destroy_guard_condition(gc); +} diff --git a/rmw_unix_socket_cpp/test/test_rmw_service_client.cpp b/rmw_unix_socket_cpp/test/test_rmw_service_client.cpp index d99e2b0..227d00e 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_service_client.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_service_client.cpp @@ -21,8 +21,6 @@ #include "rmw/qos_profiles.h" #include "rosidl_typesupport_cpp/service_type_support.hpp" -#include "../src/types.hpp" // UdsService/UdsClient shm_ring layout only (no linked symbols) - class ServiceClientTest : public RmwUdsNodeTest { protected: @@ -136,12 +134,10 @@ TEST_F(ServiceClientTest, RequestResponseRoundTrip) EXPECT_EQ("success", recv_response.string_value); } -TEST_F(ServiceClientTest, LargeRequestAndResponseViaShm) +TEST_F(ServiceClientTest, LargeRequestAndResponseRoundTrip) { - // A >4 MB request and a >4 MB response must round-trip. Both exceed the - // kernel's ~4 MB single-datagram cap (buffer-independent), so success proves - // the client's and service's shm rings carried the payloads: on the inline - // path (pre-change) the datagrams would be rejected and never delivered. + // A >4 MB request and a >4 MB response (both beyond the kernel's ~4 MB + // single-datagram cap) must round-trip byte-for-byte. srv = rmw_create_service(node, ts, "/large_srv", &qos); cli = rmw_create_client(node, ts, "/large_srv", &qos); ASSERT_NE(nullptr, srv); @@ -162,10 +158,6 @@ TEST_F(ServiceClientTest, LargeRequestAndResponseViaShm) int64_t seq_id = 0; EXPECT_EQ(RMW_RET_OK, rmw_send_request(cli, &request, &seq_id)); - auto * cli_data = static_cast(cli->data); - EXPECT_NE(nullptr, cli_data->shm_ring.base) - << "a >4 MB request must be staged into the client's shm ring"; - test_msgs::srv::BasicTypes::Request recv_request; rmw_service_info_t request_header; std::memset(&request_header, 0, sizeof(request_header)); @@ -180,10 +172,6 @@ TEST_F(ServiceClientTest, LargeRequestAndResponseViaShm) response.string_value = big_resp; EXPECT_EQ(RMW_RET_OK, rmw_send_response(srv, &request_header.request_id, &response)); - auto * srv_data = static_cast(srv->data); - EXPECT_NE(nullptr, srv_data->shm_ring.base) - << "a >4 MB response must be staged into the service's shm ring"; - test_msgs::srv::BasicTypes::Response recv_response; rmw_service_info_t response_header; std::memset(&response_header, 0, sizeof(response_header)); @@ -194,12 +182,10 @@ TEST_F(ServiceClientTest, LargeRequestAndResponseViaShm) EXPECT_EQ(big_resp, recv_response.string_value); } -TEST_F(ServiceClientTest, LargeRequestDeliveredThroughWaitDrain) +TEST_F(ServiceClientTest, LargeRequestDeliveredToServiceBlockedInWait) { - // rmw_wait drains service sockets into the request queue; it must resolve a - // large-request shm descriptor there too, not only in rmw_take_request. Real - // executors always wait before taking, so if the wait drain dropped the - // descriptor the request would be lost before take ever ran. + // A 5 MB request sent to a service that is blocked in rmw_wait must still be + // delivered: real executors always wait before taking. srv = rmw_create_service(node, ts, "/large_wait_srv", &qos); cli = rmw_create_client(node, ts, "/large_wait_srv", &qos); ASSERT_NE(nullptr, srv); @@ -215,7 +201,7 @@ TEST_F(ServiceClientTest, LargeRequestDeliveredThroughWaitDrain) int64_t seq_id = 0; EXPECT_EQ(RMW_RET_OK, rmw_send_request(cli, &request, &seq_id)); - // Drain via rmw_wait (not a direct take) to exercise the wait-side path. + // Wait on the service before taking, as a real executor would. auto * ws = rmw_create_wait_set(&context, 1); ASSERT_NE(nullptr, ws); rmw_services_t services; @@ -228,16 +214,16 @@ TEST_F(ServiceClientTest, LargeRequestDeliveredThroughWaitDrain) EXPECT_EQ( RMW_RET_OK, rmw_wait(nullptr, nullptr, &services, nullptr, nullptr, ws, &timeout)); + // Destroy before the ASSERTs below so a failure doesn't leak the wait set. + EXPECT_EQ(RMW_RET_OK, rmw_destroy_wait_set(ws)); test_msgs::srv::BasicTypes::Request recv_request; rmw_service_info_t request_header; std::memset(&request_header, 0, sizeof(request_header)); bool taken = false; EXPECT_EQ(RMW_RET_OK, rmw_take_request(srv, &request_header, &recv_request, &taken)); - ASSERT_TRUE(taken) << "a 5 MB request drained by rmw_wait must be delivered"; + ASSERT_TRUE(taken) << "a 5 MB request must be delivered after rmw_wait"; EXPECT_EQ(big, recv_request.string_value); - - EXPECT_EQ(RMW_RET_OK, rmw_destroy_wait_set(ws)); } TEST_F(ServiceClientTest, SendResponseToGoneClientReturnsOk) diff --git a/rmw_unix_socket_cpp/test/test_rmw_wait.cpp b/rmw_unix_socket_cpp/test/test_rmw_wait.cpp index fe86ff3..9ff16d2 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_wait.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_wait.cpp @@ -14,15 +14,16 @@ #include "test_base.hpp" +#include +#include #include +#include #include "test_msgs/msg/basic_types.hpp" #include "rmw/qos_profiles.h" #include "rosidl_typesupport_cpp/message_type_support.hpp" -#include "../src/types.hpp" - TEST_F(RmwUdsTestBase, CreateDestroyWaitSet) { auto * ws = rmw_create_wait_set(&context, 10); @@ -139,3 +140,174 @@ TEST_F(RmwUdsNodeTest, WaitWithSubscription) EXPECT_EQ(RMW_RET_OK, rmw_destroy_subscription(node, sub)); EXPECT_EQ(RMW_RET_OK, rmw_destroy_publisher(node, pub)); } + +TEST_F(RmwUdsNodeTest, NodeGraphGuardConditionTriggersOnGraphChange) +{ + // Scenario: graph-change notification. rclcpp's GraphListener (and + // wait_for_service, on_graph_change callbacks) blocks on the node's graph + // guard condition and relies on it firing when the ROS graph changes. + // Block on that guard condition alone, then create a subscription from the + // main thread: the wait must wake with the guard condition ready, well + // before the timeout. Without this, wait_for_service can hang forever even + // though the service is up. + const rmw_guard_condition_t * graph_gc = rmw_node_get_graph_guard_condition(node); + ASSERT_NE(nullptr, graph_gc); + + auto * ws = rmw_create_wait_set(&context, 1); + ASSERT_NE(nullptr, ws); + + std::atomic woke_ready{false}; + std::thread waiter( + [&] { + void * gc_array[1] = {graph_gc->data}; + rmw_guard_conditions_t gcs; + gcs.guard_conditions = gc_array; + gcs.guard_condition_count = 1; + rmw_time_t timeout{3, 0}; + rmw_ret_t ret = rmw_wait(nullptr, &gcs, nullptr, nullptr, nullptr, ws, &timeout); + // Ready iff rmw_wait kept the entry non-null and returned OK. + woke_ready.store(ret == RMW_RET_OK && gcs.guard_conditions[0] != nullptr); + }); + + // Let the waiter reach epoll, then change the graph. + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + auto * ts_local = rosidl_typesupport_cpp::get_message_type_support_handle< + test_msgs::msg::BasicTypes>(); + rmw_qos_profile_t qos = rmw_qos_profile_default; + auto sub_opts = rmw_get_default_subscription_options(); + auto * sub = rmw_create_subscription(node, ts_local, "/graph_gc_probe", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + + waiter.join(); + EXPECT_TRUE(woke_ready.load()) << + "the node's graph guard condition was not triggered by a graph change"; + + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_wait_set(ws); +} + +TEST_F(RmwUdsNodeTest, WaitBlocksForFullCallerTimeout) +{ + // Scenario: the caller's timeout is a contract. Callers such as + // rclcpp::wait_for_message and WaitSet::wait treat an early RMW_RET_TIMEOUT + // as "nothing arrived in my window" — if rmw_wait returns before the + // caller's deadline, they misreport. With nothing ready, a 600 ms wait must + // block ~600 ms and only then return RMW_RET_TIMEOUT. + auto * gc = rmw_create_guard_condition(&context); + ASSERT_NE(nullptr, gc); + auto * ws = rmw_create_wait_set(&context, 1); + ASSERT_NE(nullptr, ws); + + void * gc_array[1] = {gc->data}; + rmw_guard_conditions_t gcs; + gcs.guard_conditions = gc_array; + gcs.guard_condition_count = 1; + + rmw_time_t timeout{0, 600000000}; // 600 ms, never triggered + auto t0 = std::chrono::steady_clock::now(); + rmw_ret_t ret = rmw_wait(nullptr, &gcs, nullptr, nullptr, nullptr, ws, &timeout); + auto elapsed_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0).count(); + + EXPECT_EQ(RMW_RET_TIMEOUT, ret); + EXPECT_GE(elapsed_ms, 550) << + "rmw_wait returned TIMEOUT before the caller's 600 ms deadline"; + EXPECT_LE(elapsed_ms, 1500) << "rmw_wait overshot the deadline"; + + auto _r1 [[maybe_unused]] = rmw_destroy_wait_set(ws); + auto _r2 [[maybe_unused]] = rmw_destroy_guard_condition(gc); +} + +TEST_F(RmwUdsNodeTest, LatchedTopicSurvivesAnUnresponsiveParticipant) +{ + // Scenario: one participant on the domain initializes but never services + // its wait loop (a hung or busy process). However much graph churn its + // unread notifications accumulate, the rest of the system must keep + // working: a latched (TRANSIENT_LOCAL) message published by an idle node + // must still reach a subscriber that joins after heavy churn. + rmw_init_options_t opts2 = rmw_get_zero_initialized_init_options(); + rcutils_allocator_t allocator = rcutils_get_default_allocator(); + ASSERT_EQ(RMW_RET_OK, rmw_init_options_init(&opts2, allocator)); + opts2.domain_id = 99; // same domain as the fixture + rmw_context_t ctx2 = rmw_get_zero_initialized_context(); + ASSERT_EQ(RMW_RET_OK, rmw_init(&opts2, &ctx2)); // never waits, never drains + + auto * ts_local = rosidl_typesupport_cpp::get_message_type_support_handle< + test_msgs::msg::BasicTypes>(); + rmw_qos_profile_t latched = rmw_qos_profile_default; + latched.reliability = RMW_QOS_POLICY_RELIABILITY_RELIABLE; + latched.durability = RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL; + latched.depth = 5; + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher( + node, ts_local, "/unresponsive_latched", &latched, &pub_opts); + ASSERT_NE(nullptr, pub); + test_msgs::msg::BasicTypes m; + m.int32_value = 21; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + + // The healthy participant's executor: idle, blocked, servicing its waits — + // exactly what a quiet production node does. + auto * gc = rmw_create_guard_condition(&context); + ASSERT_NE(nullptr, gc); + auto * ws = rmw_create_wait_set(&context, 4); + ASSERT_NE(nullptr, ws); + std::atomic stop{false}; + std::thread executor( + [&] { + while (!stop.load()) { + void * gc_array[1] = {gc->data}; + rmw_guard_conditions_t gcs; + gcs.guard_conditions = gc_array; + gcs.guard_condition_count = 1; + auto _r [[maybe_unused]] = rmw_wait( + nullptr, &gcs, nullptr, nullptr, nullptr, ws, nullptr); + } + }); + std::this_thread::sleep_for(std::chrono::milliseconds(150)); + + // Heavy graph churn while the second participant stays unresponsive. The + // healthy executor keeps draining its own notifications throughout, so any + // per-sender resource pinned by the unresponsive peer stays pinned. + rmw_qos_profile_t qos = rmw_qos_profile_default; + for (int i = 0; i < 600; ++i) { + auto * p = rmw_create_publisher(node, ts_local, "/churn", &qos, &pub_opts); + ASSERT_NE(nullptr, p); + ASSERT_EQ(RMW_RET_OK, rmw_destroy_publisher(node, p)); + } + + // A late joiner after the churn must still receive the latched message. + auto sub_opts = rmw_get_default_subscription_options(); + auto * sub = rmw_create_subscription( + node, ts_local, "/unresponsive_latched", &latched, &sub_opts); + ASSERT_NE(nullptr, sub); + + bool got = false; + for (int i = 0; i < 300 && !got; ++i) { + test_msgs::msg::BasicTypes recv; + bool taken = false; + if (rmw_take(sub, &recv, &taken, nullptr) == RMW_RET_OK && taken && + recv.int32_value == 21) + { + got = true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + stop.store(true); + auto _t [[maybe_unused]] = rmw_trigger_guard_condition(gc); + executor.join(); + + EXPECT_TRUE(got) << + "a participant that never drains its notifications starved a healthy " + "idle publisher: the latched message never reached the late joiner"; + + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_wait_set(ws); + auto _r3 [[maybe_unused]] = rmw_destroy_publisher(node, pub); + auto _r4 [[maybe_unused]] = rmw_destroy_guard_condition(gc); + EXPECT_EQ(RMW_RET_OK, rmw_shutdown(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_context_fini(&ctx2)); + EXPECT_EQ(RMW_RET_OK, rmw_init_options_fini(&opts2)); +}