Skip to content

Commit 6d9793c

Browse files
fix(registry): don't free a slot before its teardown finishes (#81)
registry_remove and registry_cleanup_stale CASed the slot state to EMPTY and only then ran teardown_slot. In that gap another process's try_add_once could claim the slot and write its entry, which the late teardown then zeroed (and whose socket file it unlinked). The live endpoint vanished from the graph and a nameless ghost stayed behind: state=SERVICE, pid 0, empty name and type, which cleanup_stale never reclaims. rmw_get_service_names_and_types then returned ('', ['']) and foxglove_bridge aborted ("Topic or service name can't be empty"). Seen in a 10-robot fleet sharing one PID namespace, where many processes sweep the same dead spawner PIDs while others register at launch. Hold the slot in ENTRY_RESERVED during teardown and store ENTRY_EMPTY after it. Adders only claim EMPTY and readers already skip RESERVED, so the shm layout is unchanged. registry_remove also returns early on RESERVED. Test: StaleReapNeverClobbersConcurrentAdd fails on main (~1,400 of 20,000 live entries wiped per run, 3/3 runs) and passes with the fix.
1 parent 330b421 commit 6d9793c

4 files changed

Lines changed: 98 additions & 17 deletions

File tree

‎rmw_unix_socket_cpp/DESIGN.md‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -394,7 +394,7 @@ The release store pairs with the acquire load every reader does on `state`. Any
394394

395395
**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.
396396

397-
**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.
397+
**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, CASes its `state` to `ENTRY_RESERVED`, tears it down, and only then stores `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.
398398

399399
**Failure path.** If the scan finds no empty slot, `registry_add` runs the stale-cleanup pass once (reclaiming slots whose owning PID is dead) and retries the scan a single time. If the table is still full after that, the add fails and returns -1. There is no resize and no spinning.
400400

@@ -436,9 +436,9 @@ Services and clients cache the same way against the same generation counter; the
436436

437437
Two failure modes leave entries behind: a clean shutdown, and a process that dies without running its destructors. They are handled separately because only one of them can be trusted to tell the truth about itself.
438438

439-
**Clean shutdown.** When an endpoint is destroyed normally, it removes its own slot. Because each entity records the slot index it was assigned at registration, removal needs no scan: it indexes straight to its slot. `registry_remove` reads the slot's current `state`, then attempts a single compare-and-swap from that state value to `ENTRY_EMPTY`. The CAS is what makes the removal safe under concurrency: exactly one caller can win it, and the winner owns the teardown. If the CAS fails, someone else already emptied the slot, so the function returns without touching anything.
439+
**Clean shutdown.** When an endpoint is destroyed normally, it removes its own slot. Because each entity records the slot index it was assigned at registration, removal needs no scan: it indexes straight to its slot. `registry_remove` reads the slot's current `state`, then attempts a single compare-and-swap from that state value to `ENTRY_RESERVED`. The CAS is what makes the removal safe under concurrency: exactly one caller can win it, and the winner owns the teardown. If the CAS fails (or the slot is already `ENTRY_EMPTY` or `ENTRY_RESERVED`), someone else is emptying it, so the function returns without touching anything.
440440

441-
The teardown itself runs under the slot's seqlock (the same odd/even protocol used on the write side). `teardown_slot` bumps `seq` to odd, copies out the socket path, zeroes every payload field, then bumps `seq` back to even. Any reader that was mid-snapshot sees the odd `seq` and retries, so it can never read a half-erased slot. The socket file is unlinked after the seqlock closes, because a filesystem operation does not need to be observable atomically by other processes. Finally `generation` is incremented so every cached reader learns the graph changed. The order matters: `state` flips to `ENTRY_EMPTY` first (no new reader can match the slot), the payload is wiped under the seqlock, and only then does `generation` move.
441+
The teardown itself runs under the slot's seqlock (the same odd/even protocol used on the write side). `teardown_slot` bumps `seq` to odd, copies out the socket path, zeroes every payload field, then bumps `seq` back to even. Any reader that was mid-snapshot sees the odd `seq` and retries, so it can never read a half-erased slot. The socket file is unlinked after the seqlock closes, because a filesystem operation does not need to be observable atomically by other processes. Finally `generation` is incremented so every cached reader learns the graph changed. The order matters: `state` flips to `ENTRY_RESERVED` first (no new reader can match the slot, and no adder can claim it), the payload is wiped under the seqlock, `state` is set to `ENTRY_EMPTY`, and only then does `generation` move. Freeing the slot before the wipe would let a concurrent add claim it in between and have its fresh entry and socket file erased by the late teardown.
442442

443443
**Crash.** A process killed by SIGKILL, OOM, or a container teardown never reaches `registry_remove`. Its slot keeps a stale socket path that no longer accepts connections. There is no destructor to trust, so the registry infers death from the operating system instead. `registry_cleanup_stale` walks the table, and for each live slot it takes a seqlock snapshot and checks the owning PID by stat-ing `/proc/<pid>`. The decision is deliberately conservative:
444444

@@ -448,7 +448,7 @@ The teardown itself runs under the slot's seqlock (the same odd/even protocol us
448448

449449
The bias toward keeping is intentional. The cost of wrongly keeping a dead slot is one wasted slot until the next pass; the cost of wrongly reclaiming a live slot is tearing down a working endpoint. A `pid` of 0 is also never reclaimed, since that value only appears in the transient snapshot of a slot whose payload has not been written yet, never in a committed entry.
450450

451-
The dangerous window is between vetting a slot and tearing it down. The owner could exit, the slot could be reclaimed by a clean remove, and a different process could re-add an entry of the same type into the same slot, all before this pass acts. This is the classic ABA problem: the slot looks identical but is a different entry, and the `state` value alone cannot tell them apart. The seqlock counter closes the gap. The snapshot captures the slot's `seq` (`snap_seq`), and immediately before the reclaim CAS the pass re-reads `seq` and compares. Any intervening write (a remove or a re-add both go through the seqlock) advances the counter, so a mismatch means the slot is no longer the entry that was checked, and the pass skips it. Only when `seq` is unchanged does it CAS the observed `state` to `ENTRY_EMPTY`, log a one-line breadcrumb naming the slot, PID, node, and topic so an ungraceful exit is visible in the ROS log, run the same `teardown_slot`, and bump `generation`.
451+
The dangerous window is between vetting a slot and tearing it down. The owner could exit, the slot could be reclaimed by a clean remove, and a different process could re-add an entry of the same type into the same slot, all before this pass acts. This is the classic ABA problem: the slot looks identical but is a different entry, and the `state` value alone cannot tell them apart. The seqlock counter closes the gap. The snapshot captures the slot's `seq` (`snap_seq`), and immediately before the reclaim CAS the pass re-reads `seq` and compares. Any intervening write (a remove or a re-add both go through the seqlock) advances the counter, so a mismatch means the slot is no longer the entry that was checked, and the pass skips it. Only when `seq` is unchanged does it CAS the observed `state` to `ENTRY_RESERVED`, log a one-line breadcrumb naming the slot, PID, node, and topic so an ungraceful exit is visible in the ROS log, run the same `teardown_slot`, store `ENTRY_EMPTY`, and bump `generation`.
452452

453453
**When it runs.** There are two triggers.
454454

@@ -476,7 +476,7 @@ So the hot paths avoid locks entirely. They rely on the fact that `std::atomic<u
476476

477477
Three mechanisms replace the lock:
478478

479-
- **Claiming a slot is a single CAS.** To add an entry, a writer scans for an empty slot and flips its `state` from `ENTRY_EMPTY` to `ENTRY_RESERVED` with one atomic compare-and-swap. Exactly one writer wins. `ENTRY_RESERVED` is a private "claimed but not yet filled in" marker (see Registering an entry for the full read/write protocol). Removal is the mirror image: a CAS of `state` back to `ENTRY_EMPTY` decides who owns the teardown.
479+
- **Claiming a slot is a single CAS.** To add an entry, a writer scans for an empty slot and flips its `state` from `ENTRY_EMPTY` to `ENTRY_RESERVED` with one atomic compare-and-swap. Exactly one writer wins. `ENTRY_RESERVED` is a private "claimed but not yet filled in" marker (see Registering an entry for the full read/write protocol). Removal is the mirror image: a CAS of `state` to `ENTRY_RESERVED` decides who owns the teardown, and `ENTRY_EMPTY` is stored only once the slot is wiped.
480480

481481
- **Publishing and reading a slot is a seqlock.** A writer brackets the payload copy with an odd/even `seq` counter; a reader snapshots `seq` before and after the copy and retries on mismatch. The full protocol is described in Registering an entry. The point for locking is that readers never block a writer and writers never block a reader, and the bounded retry (`MAX_READ_RETRIES`) guarantees a persistent writer cannot spin a reader forever.
482482

‎rmw_unix_socket_cpp/src/registry.cpp‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -305,10 +305,11 @@ int32_t registry_add(RegistryHeader * header, const RegistryEntry & entry)
305305
return try_add_once(header, entry);
306306
}
307307

308-
// Best-effort tear-down of a slot we already claimed (state already EMPTY).
309-
// Unlink the socket file and zero out the payload so the next reuse starts
310-
// from a clean state. Wrapped in a seqlock so any reader still in the middle
311-
// of a snapshot retries.
308+
// Best-effort tear-down of a slot we already claimed (state RESERVED). Unlink the
309+
// socket file and zero out the payload so the next reuse starts from a clean
310+
// state. Wrapped in a seqlock so any reader still in the middle of a snapshot
311+
// retries. The caller stores EMPTY only afterwards: freeing the slot first would
312+
// let a concurrent add claim it and have its fresh entry (and socket) wiped here.
312313
static void teardown_slot(RegistryEntrySlot * slot)
313314
{
314315
slot->seq.fetch_add(1, std::memory_order_acq_rel);
@@ -352,17 +353,19 @@ void registry_remove(RegistryHeader * header, int32_t index)
352353
auto * slot = &registry_slots(header)[index];
353354

354355
uint8_t st = slot->state.load(std::memory_order_acquire);
355-
if (st == ENTRY_EMPTY) {
356-
return;
356+
if (st == ENTRY_EMPTY || st == ENTRY_RESERVED) {
357+
return; // already free, or another thread is mid-add/mid-teardown
357358
}
358-
// CAS state -> EMPTY. Winner owns the teardown.
359+
// CAS state -> RESERVED. Winner owns the teardown; the slot stays unclaimable
360+
// until the payload is zeroed.
359361
if (!slot->state.compare_exchange_strong(
360-
st, static_cast<uint8_t>(ENTRY_EMPTY),
362+
st, static_cast<uint8_t>(ENTRY_RESERVED),
361363
std::memory_order_acq_rel, std::memory_order_relaxed))
362364
{
363365
return; // someone else removed it
364366
}
365367
teardown_slot(slot);
368+
slot->state.store(static_cast<uint8_t>(ENTRY_EMPTY), std::memory_order_release);
366369
header->generation.fetch_add(1, std::memory_order_acq_rel);
367370
if (st != ENTRY_DOORBELL) { // doorbell slots have no wake consumers
368371
ring_doorbells(header); // strictly after the bump — see ring_doorbells
@@ -518,10 +521,11 @@ void registry_cleanup_stale(RegistryHeader * header)
518521
continue; // slot was rewritten since snapshot — not the entry we vetted
519522
}
520523
// Try to claim the slot for removal. The CAS-expected value is the state
521-
// we observed in the snapshot.
524+
// we observed in the snapshot. RESERVED (not EMPTY) keeps adders out until
525+
// teardown_slot has zeroed it; see teardown_slot.
522526
uint8_t expected = snap.state.load(std::memory_order_relaxed);
523527
if (!slots[i].state.compare_exchange_strong(
524-
expected, static_cast<uint8_t>(ENTRY_EMPTY),
528+
expected, static_cast<uint8_t>(ENTRY_RESERVED),
525529
std::memory_order_acq_rel, std::memory_order_relaxed))
526530
{
527531
continue; // raced with another remover / writer
@@ -539,6 +543,7 @@ void registry_cleanup_stale(RegistryHeader * header)
539543
snap.topic_name[0] ? snap.topic_name : "(none)");
540544

541545
teardown_slot(&slots[i]);
546+
slots[i].state.store(static_cast<uint8_t>(ENTRY_EMPTY), std::memory_order_release);
542547
header->generation.fetch_add(1, std::memory_order_acq_rel);
543548
if (expected != static_cast<uint8_t>(ENTRY_DOORBELL)) {
544549
reclaimed = true; // doorbell-only reclaims ring nobody

‎rmw_unix_socket_cpp/src/registry.hpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,11 +104,12 @@ static_assert(offsetof(RegistryEntry, qos_depth) == 1160, "RegistryEntry layout
104104
// 4. s2 = seq.load(acquire); if (s1 != s2) retry (bounded retries, then skip)
105105
//
106106
// Protocol (remove):
107-
// 1. CAS state <current> -> EMPTY (claims removal; loses race => abort)
107+
// 1. CAS state <current> -> RESERVED (claims removal; loses race => abort)
108108
// 2. seq.fetch_add(1) -> begin payload teardown
109109
// 3. unlink(socket_path); memset payload
110110
// 4. seq.fetch_add(1) -> end teardown
111-
// 5. generation.fetch_add(1)
111+
// 5. state.store(EMPTY, release) -> only now can an adder claim the slot
112+
// 6. generation.fetch_add(1)
112113
//
113114
// All atomics are lock-free + address-free (std::atomic<u8/u32>::is_always_lock_free
114115
// is true on every Linux target we support), so they work across process

‎rmw_unix_socket_cpp/test/test_registry_concurrent.cpp‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
#include <vector>
3939

4040
#include <sys/mman.h>
41+
#include <sys/wait.h>
4142
#include <unistd.h>
4243

4344
#include "../src/registry.hpp"
@@ -340,3 +341,77 @@ TEST_F(RegistryConcurrentTest, GenerationCounterMonotonicUnderChaos)
340341
EXPECT_EQ(0, regressions.load())
341342
<< "generation counter regressed " << regressions.load() << " times";
342343
}
344+
345+
// A stale-PID reap must never tear down an entry that a live process wrote into
346+
// the same slot after the reaper claimed it. Before the fix the reaper freed the
347+
// slot (state -> EMPTY) before zeroing it, so a concurrent add could take the
348+
// slot and have its payload wiped: the live entry vanished and a nameless
349+
// "ghost" stayed behind (seen as an empty service name in a 10-robot fleet).
350+
TEST_F(RegistryConcurrentTest, StaleReapNeverClobbersConcurrentAdd)
351+
{
352+
auto * header = rmw_uds::registry_header(registry_ptr);
353+
354+
pid_t dead = fork();
355+
ASSERT_GE(dead, 0);
356+
if (dead == 0) {_exit(0);}
357+
ASSERT_EQ(dead, waitpid(dead, nullptr, 0)); // reaped: /proc/<dead> is gone
358+
359+
std::atomic<bool> stop{false};
360+
std::atomic<int> lost{0}, ghosts{0};
361+
362+
// Mimics a launch spawner that exits uncleanly: its entry keeps landing in
363+
// the lowest free slot and a sweep keeps reaping it.
364+
std::thread reaper([&]() {
365+
while (!stop.load()) {
366+
rmw_uds::RegistryEntry e;
367+
std::memset(&e, 0, sizeof(e));
368+
e.type = rmw_uds::ENTRY_SERVICE;
369+
e.pid = dead;
370+
std::snprintf(e.topic_name, sizeof(e.topic_name), "/dead_srv");
371+
std::snprintf(e.type_name, sizeof(e.type_name), "T");
372+
rmw_uds::registry_add(header, e);
373+
rmw_uds::registry_cleanup_stale(header);
374+
}
375+
});
376+
377+
std::thread detector([&]() {
378+
while (!stop.load()) {
379+
for (auto & r : rmw_uds::registry_query(
380+
header, rmw_uds::ENTRY_SERVICE, nullptr, nullptr, nullptr))
381+
{
382+
if (r.topic_name.empty()) {ghosts.fetch_add(1);}
383+
}
384+
}
385+
});
386+
387+
std::vector<std::thread> writers;
388+
for (int t = 0; t < 4; ++t) {
389+
writers.emplace_back([&, t]() {
390+
for (int k = 0; k < 5000; ++k) {
391+
rmw_uds::RegistryEntry e;
392+
std::memset(&e, 0, sizeof(e));
393+
e.type = rmw_uds::ENTRY_SERVICE;
394+
e.pid = getpid();
395+
std::snprintf(e.topic_name, sizeof(e.topic_name), "/live_%d_%d", t, k);
396+
std::snprintf(e.type_name, sizeof(e.type_name), "T");
397+
int32_t idx = rmw_uds::registry_add(header, e);
398+
ASSERT_GE(idx, 0);
399+
// Give an in-flight reap of this slot time to finish its teardown.
400+
std::this_thread::sleep_for(std::chrono::microseconds(50));
401+
if (rmw_uds::registry_query(
402+
header, rmw_uds::ENTRY_SERVICE, e.topic_name, nullptr, nullptr).size() != 1)
403+
{
404+
lost.fetch_add(1);
405+
}
406+
rmw_uds::registry_remove(header, idx);
407+
}
408+
});
409+
}
410+
for (auto & w : writers) {w.join();}
411+
stop.store(true);
412+
reaper.join();
413+
detector.join();
414+
415+
EXPECT_EQ(0, lost.load()) << "live entries wiped by a concurrent stale reap";
416+
EXPECT_EQ(0, ghosts.load()) << "nameless entries observed after a reap clobbered an add";
417+
}

0 commit comments

Comments
 (0)