perf(registry): ring doorbells from a cache instead of scanning every…

rmw_unix_socket_cpp/src/registry.cpp

@@ -18,6 +18,7 @@
 #include <cstdio>
 #include <cstring>
 #include <sched.h>
+#include <vector>
 
 #include <fcntl.h>
 #include <sys/mman.h>
@@ -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<int32_t>(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<const struct sockaddr *>(&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<const struct sockaddr *>(&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<struct sockaddr_un> 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<const struct sockaddr *>(&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<const struct sockaddr *>(&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;
   }

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<uint32_t> 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<uint32_t> 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.

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 <sys/socket.h>
 #include <sys/stat.h>
 
 #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;

rmw_unix_socket_cpp/test/test_registry.cpp

@@ -19,7 +19,9 @@
 #include <fcntl.h>
 #include <unistd.h>
 #include <sys/mman.h>
+#include <sys/socket.h>
 #include <sys/stat.h>
+#include <sys/un.h>
 
 #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<struct sockaddr *>(&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);