From f2be7d3ec340528ac83d60fa170c083fd26bb932 Mon Sep 17 00:00:00 2001 From: Abderahmane BENALI Date: Tue, 11 Aug 2026 18:04:25 +0200 Subject: [PATCH] perf(registry): ring doorbells from a cache instead of scanning every slot ring_doorbells runs after every registry mutation and walked [0, high_water) looking for doorbell slots. On a busy domain that is thousands of strided slot reads, each its own cache line, to find the handful of entries that are doorbells. Creating or destroying any node, publisher, subscription, service or client paid it. Doorbells only come and go when a process starts or stops, which is far rarer than entity churn, so add doorbell_generation to the header and bump it only on an ENTRY_DOORBELL claim or release. Each ringing thread caches the resolved doorbell addresses and rebuilds only when that counter moves, so the common mutation rings straight from the vector with no scan at all. doorbell_generation lands in the padding that already followed high_water_slot: sizeof(RegistryHeader) stays 64, slots still start at 64, and the total shm size is unchanged. A static_assert pins it there. REGISTRY_VERSION still goes to 5, because a v4 writer never bumps the counter and would leave a v5 reader holding a stale cache and missing wakeups, so the two must not share a domain. Also shrink the doorbell socket's receive buffer. create_bound_socket sizes buffers for real payloads, but a doorbell datagram carries none: it only says "the registry changed", and rmw_wait drains the whole queue and re-reads the registry once. Bounding the queue lets the kernel coalesce a storm instead of piling up hundreds of datagrams for one re-read. Removing the close-and-retry on EAGAIN was tried and reverted: LatchedTopicSurvivesAnUnresponsiveParticipant fails without it. AF_UNIX datagrams stay charged to the sender's buffer until the receiver consumes them, so a peer that never drains still starves sends to healthy peers. A smaller receive buffer reduces how much one peer can pin but does not remove the effect, so the reset stays, now shared by both send paths in ring_one(). Closes #53 --- rmw_unix_socket_cpp/src/registry.cpp | 106 ++++++++++++++++----- rmw_unix_socket_cpp/src/registry.hpp | 20 +++- rmw_unix_socket_cpp/src/rmw_init.cpp | 19 ++++ rmw_unix_socket_cpp/test/test_registry.cpp | 87 +++++++++++++++++ 4 files changed, 206 insertions(+), 26 deletions(-) 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);