diff --git a/rmw_unix_socket_cpp/src/registry.cpp b/rmw_unix_socket_cpp/src/registry.cpp index 84080b3..7eb5788 100644 --- a/rmw_unix_socket_cpp/src/registry.cpp +++ b/rmw_unix_socket_cpp/src/registry.cpp @@ -18,6 +18,7 @@ #include #include #include +#include #include #include @@ -68,6 +69,7 @@ static void initialize_header(RegistryHeader * header, size_t total_size) header->max_entries = REGISTRY_MAX_ENTRIES; header->generation.store(0, std::memory_order_relaxed); header->high_water_slot.store(0, std::memory_order_relaxed); + header->doorbell_generation.store(0, std::memory_order_relaxed); pthread_mutexattr_t attr; pthread_mutexattr_init(&attr); @@ -280,6 +282,13 @@ static int32_t try_add_once(RegistryHeader * header, const RegistryEntry & entry while (cur < want && !header->high_water_slot.compare_exchange_weak( cur, want, std::memory_order_relaxed, std::memory_order_relaxed)) {} + // Same ordering argument as high_water_slot: sequenced before the release + // generation bump, so a ringer that synchronizes on the generation also + // sees that its cached doorbell list is stale and rebuilds it. Getting + // this backwards would let a just-started process miss wakeups. + if (entry.type == ENTRY_DOORBELL) { + header->doorbell_generation.fetch_add(1, 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); @@ -350,6 +359,9 @@ void registry_remove(RegistryHeader * header, int32_t index) return; // someone else removed it } teardown_slot(slot); + if (st == ENTRY_DOORBELL) { + header->doorbell_generation.fetch_add(1, std::memory_order_relaxed); + } header->generation.fetch_add(1, std::memory_order_acq_rel); ring_doorbells(header); // strictly after the bump — see ring_doorbells } @@ -392,26 +404,77 @@ static const char * entry_type_name(uint8_t t) // 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). +// Send one wake octet, best-effort. EAGAIN gets one retry on a fresh fd: +// AF_UNIX datagrams stay charged to the SENDER's buffer until the receiver +// consumes them, so a peer that never drains can exhaust this fd's budget and +// make sendto fail for every OTHER peer too. Resetting the fd releases that +// budget so one hung participant cannot mute the whole domain. Bounding the +// receiver's queue (see the doorbell SO_RCVBUF in rmw_init) shrinks how much +// any one peer can pin, but does not remove the effect, so the reset stays. +// Regression cover: LatchedTopicSurvivesAnUnresponsiveParticipant. +static void ring_one(int & fd, const struct sockaddr_un & addr) +{ + const uint8_t octet = 1; + const ssize_t sent = sendto( + fd, &octet, 1, MSG_DONTWAIT | MSG_NOSIGNAL, + reinterpret_cast(&addr), sizeof(addr)); + if (sent >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK)) { + return; // Delivered, or the peer is gone and its slot will be reclaimed. + } + close(fd); + fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + if (fd < 0) { + return; + } + // EAGAIN again on a fresh fd means this peer's own queue is full, so a + // wakeup is already pending there and dropping this one is safe. + (void)sendto( + fd, &octet, 1, MSG_DONTWAIT | MSG_NOSIGNAL, + reinterpret_cast(&addr), sizeof(addr)); +} + 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 + // One ring socket per mutating thread, closed at thread exit, plus the + // doorbell addresses this thread last resolved. Rebuilding that list means + // scanning every used slot to find the handful that are doorbells, so it is + // keyed on doorbell_generation, which only moves when a process registers or + // drops its doorbell. Entity churn therefore rings straight from the vector. + struct RingState { int fd = -1; - ~RingFd() {if (fd >= 0) {close(fd);}} + uint32_t built_at = 0; + bool primed = false; + std::vector addrs; + ~RingState() {if (fd >= 0) {close(fd);}} }; - static thread_local RingFd ring; + static thread_local RingState ring; if (ring.fd < 0) { ring.fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); if (ring.fd < 0) { return; } } + + // Read the doorbell generation BEFORE the scan. If a doorbell is registered + // during the scan we may miss it here, but then this load happened before + // that registration, which happened before the new process reads the + // generation it is about to block on — so our own generation bump, which + // already happened before this call, is visible in that read. The wakeup is + // carried by the generation instead of the datagram, and the next ring + // rebuilds. No lost wakeup either way. + const uint32_t dgen = header->doorbell_generation.load(std::memory_order_acquire); + if (ring.primed && ring.built_at == dgen) { + for (const auto & cached : ring.addrs) { + if (ring.fd < 0) { + return; + } + ring_one(ring.fd, cached); + } + return; + } + + ring.addrs.clear(); 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) { @@ -445,24 +508,14 @@ static void ring_doorbells(RegistryHeader * header) // 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)); + ring.addrs.push_back(addr); + if (ring.fd < 0) { + return; } + ring_one(ring.fd, addr); } + ring.built_at = dgen; + ring.primed = true; } void registry_cleanup_stale(RegistryHeader * header) @@ -524,6 +577,9 @@ void registry_cleanup_stale(RegistryHeader * header) snap.topic_name[0] ? snap.topic_name : "(none)"); teardown_slot(&slots[i]); + if (expected == ENTRY_DOORBELL) { + header->doorbell_generation.fetch_add(1, std::memory_order_relaxed); + } header->generation.fetch_add(1, std::memory_order_acq_rel); reclaimed = true; } diff --git a/rmw_unix_socket_cpp/src/registry.hpp b/rmw_unix_socket_cpp/src/registry.hpp index 25c4b9d..7670281 100644 --- a/rmw_unix_socket_cpp/src/registry.hpp +++ b/rmw_unix_socket_cpp/src/registry.hpp @@ -34,7 +34,11 @@ static constexpr uint32_t REGISTRY_MAX_ENTRIES = 32768; // Layout v3: per-slot seqlock + atomic state. No global mutex on hot paths. // Layout v4: adds a monotonic high_water_slot bound to the header so query and // cleanup scan only the used slot prefix instead of all REGISTRY_MAX_ENTRIES. -static constexpr uint32_t REGISTRY_VERSION = 4; +// Layout v5: adds doorbell_generation. Same size and same slot offset as v4 +// (it lands in existing tail padding), but a v4 writer never bumps it, which +// would let a v5 reader keep a stale doorbell cache and miss wakeups. The +// version tag keeps the two apart, so upgrade a domain all at once. +static constexpr uint32_t REGISTRY_VERSION = 5; enum RegistryEntryType : uint8_t { @@ -156,6 +160,13 @@ struct RegistryHeader // query/cleanup scan only [0, high_water_slot) instead of all max_entries. // Only grows; over-scan is always safe, so reads need no strict ordering. std::atomic high_water_slot; + // Bumped only when an ENTRY_DOORBELL slot is claimed or released, which + // happens once per process rather than once per entity. ring_doorbells caches + // the doorbell addresses and rebuilds that cache only when this changes, so + // the common mutation rings straight from the cache instead of scanning every + // slot. Lands in the tail padding after high_water_slot, so sizeof and the + // slot offset are unchanged (see the offset guard below). + std::atomic doorbell_generation; }; // Header layout guard: only the integer prefix is asserted. The total size @@ -165,6 +176,13 @@ static_assert(offsetof(RegistryHeader, version) == 0, "RegistryHeader layout cha static_assert(offsetof(RegistryHeader, max_entries) == 4, "RegistryHeader layout changed"); static_assert(offsetof(RegistryHeader, generation) == 8, "RegistryHeader layout changed"); static_assert(offsetof(RegistryHeader, init_mutex) == 16, "RegistryHeader layout changed"); +// doorbell_generation must sit in the padding that already followed +// high_water_slot. If this ever moves, every slot moves with it and the shm +// becomes incompatible with the same REGISTRY_VERSION. +static_assert( + offsetof(RegistryHeader, doorbell_generation) == + offsetof(RegistryHeader, high_water_slot) + sizeof(uint32_t), + "doorbell_generation must pack into the tail padding after high_water_slot"); // Open or create the shared memory registry for the given domain_id. // Returns fd >= 0 on success, sets *out_ptr and *out_size. diff --git a/rmw_unix_socket_cpp/src/rmw_init.cpp b/rmw_unix_socket_cpp/src/rmw_init.cpp index 3e27770..9841457 100644 --- a/rmw_unix_socket_cpp/src/rmw_init.cpp +++ b/rmw_unix_socket_cpp/src/rmw_init.cpp @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +#include #include #include "identifier.hpp" @@ -209,6 +210,24 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context) RMW_SET_ERROR_MSG("failed to create doorbell socket"); return RMW_RET_ERROR; } + // Shrink the receive buffer that create_bound_socket sized for real data. + // A doorbell datagram carries no payload, it only says "the registry + // changed", so a queue of them is pure redundancy: rmw_wait drains the + // whole queue and re-reads the registry once. Bounding the queue makes the + // kernel coalesce the storm for us — a ringer hits EAGAIN and drops the + // extra ring instead of appending to a queue hundreds deep, and the drain + // side reads a couple of datagrams rather than hundreds. The kernel doubles + // this and floors it at SOCK_MIN_RCVBUF, so a handful still fit, which is + // all the no-lost-wakeup pairing needs. + const int doorbell_rcvbuf = 1024; + if (setsockopt( + ctx->doorbell_fd, SOL_SOCKET, SO_RCVBUF, + &doorbell_rcvbuf, sizeof(doorbell_rcvbuf)) != 0) + { + RMW_UDS_LOG_WARN( + "rmw_init: could not shrink doorbell SO_RCVBUF: %s — " + "wakeup datagrams may queue deeper than needed", std::strerror(errno)); + } rmw_uds::RegistryEntry dentry; std::memset(&dentry, 0, sizeof(dentry)); dentry.type = rmw_uds::ENTRY_DOORBELL; diff --git a/rmw_unix_socket_cpp/test/test_registry.cpp b/rmw_unix_socket_cpp/test/test_registry.cpp index 4fd7788..585ecf4 100644 --- a/rmw_unix_socket_cpp/test/test_registry.cpp +++ b/rmw_unix_socket_cpp/test/test_registry.cpp @@ -19,7 +19,9 @@ #include #include #include +#include #include +#include #include "../src/registry.hpp" @@ -258,6 +260,91 @@ TEST_F(RegistryTest, HighWaterBoundsScanWithoutLosingTopEntry) // which is what cleanup_stale uses to decide an entry's owner is dead. static constexpr pid_t DEAD_PID = 2147483000; +// ring_doorbells caches the doorbell addresses and only rebuilds that cache +// when doorbell_generation moves. If a newly registered doorbell failed to move +// it, every process that had already warmed its cache would stop ringing the +// new one, and that process would sit blocked through graph changes it should +// have seen. This is the regression guard for that. +TEST_F(RegistryTest, RingReachesADoorbellRegisteredAfterTheCacheWasWarmed) +{ + auto * header = rmw_uds::registry_header(registry_ptr); + + auto bind_doorbell = [&](const char * path) { + unlink(path); + int fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + if (fd < 0) {return -1;} + struct sockaddr_un addr; + std::memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + std::strncpy(addr.sun_path, path, sizeof(addr.sun_path) - 1); + if (bind(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + close(fd); + return -1; + } + return fd; + }; + auto register_doorbell = [&](const char * path) { + rmw_uds::RegistryEntry e; + std::memset(&e, 0, sizeof(e)); + e.type = rmw_uds::ENTRY_DOORBELL; + e.pid = getpid(); + std::strncpy(e.socket_path, path, sizeof(e.socket_path) - 1); + return rmw_uds::registry_add(header, e); + }; + auto drain = [](int fd) { + uint8_t buf[16]; + int n = 0; + while (recv(fd, buf, sizeof(buf), MSG_DONTWAIT) > 0) {++n;} + return n; + }; + auto add_plain_entry = [&](const char * name) { + rmw_uds::RegistryEntry e; + std::memset(&e, 0, sizeof(e)); + e.type = rmw_uds::ENTRY_NODE; + e.pid = getpid(); + std::strncpy(e.node_name, name, sizeof(e.node_name) - 1); + return rmw_uds::registry_add(header, e); + }; + + const char * path_a = "/tmp/rmw_uds_test_doorbell_a.sock"; + const char * path_b = "/tmp/rmw_uds_test_doorbell_b.sock"; + const int fd_a = bind_doorbell(path_a); + const int fd_b = bind_doorbell(path_b); + ASSERT_GE(fd_a, 0); + ASSERT_GE(fd_b, 0); + + const int32_t idx_a = register_doorbell(path_a); + ASSERT_GE(idx_a, 0); + drain(fd_a); + + // Warm this thread's cache: a plain mutation rings only A. + const int32_t idx_n1 = add_plain_entry("warmer"); + ASSERT_GE(idx_n1, 0); + EXPECT_GT(drain(fd_a), 0) << "the only registered doorbell was not rung"; + + // B joins after the cache was built, then another plain mutation. B must be + // rung, which only happens if registering it invalidated the cache. + const int32_t idx_b = register_doorbell(path_b); + ASSERT_GE(idx_b, 0); + drain(fd_a); + drain(fd_b); + + const int32_t idx_n2 = add_plain_entry("after_b"); + ASSERT_GE(idx_n2, 0); + EXPECT_GT(drain(fd_b), 0) << + "a doorbell registered after the cache was warmed never got rung"; + EXPECT_GT(drain(fd_a), 0) << "the original doorbell stopped being rung"; + + rmw_uds::registry_remove(header, idx_n1); + rmw_uds::registry_remove(header, idx_n2); + rmw_uds::registry_remove(header, idx_a); + rmw_uds::registry_remove(header, idx_b); + close(fd_a); + close(fd_b); + unlink(path_a); + unlink(path_b); +} + TEST_F(RegistryTest, CleanupStaleRemovesDeadPidEntries) { auto * header = rmw_uds::registry_header(registry_ptr);