diff --git a/rmw_unix_socket_cpp/src/registry.cpp b/rmw_unix_socket_cpp/src/registry.cpp index 01a7569..ee79649 100644 --- a/rmw_unix_socket_cpp/src/registry.cpp +++ b/rmw_unix_socket_cpp/src/registry.cpp @@ -170,7 +170,9 @@ static bool snapshot_slot(RegistryEntrySlot * slot, RegistryEntrySlot * out) continue; } uint8_t st = slot->state.load(std::memory_order_acquire); - if (st == ENTRY_EMPTY) { + if (st == ENTRY_EMPTY || st == ENTRY_RESERVED) { + // RESERVED = a writer claimed the slot but has not yet published its + // payload. Treat it as empty so we never snapshot a half-built slot. return false; } @@ -230,12 +232,17 @@ static int32_t try_add_once(RegistryHeader * header, const RegistryEntry & entry for (uint32_t i = 0; i < max; ++i) { uint8_t expected = ENTRY_EMPTY; - // CAS state EMPTY -> entry.type. Winner owns the slot. + // CAS state EMPTY -> RESERVED. Winner owns the slot, but readers still see + // it as empty until we publish the real type below — so no reader can + // snapshot the slot before its payload is committed under the seqlock. if (slots[i].state.compare_exchange_strong( - expected, static_cast(entry.type), + expected, static_cast(ENTRY_RESERVED), std::memory_order_acq_rel, std::memory_order_relaxed)) { write_slot_payload(&slots[i], entry); + // Payload is committed (seq even). Publish the real type with release + // ordering so any reader that now sees it also sees the payload. + slots[i].state.store(static_cast(entry.type), std::memory_order_release); header->generation.fetch_add(1, std::memory_order_acq_rel); return static_cast(i); } @@ -347,6 +354,11 @@ void registry_cleanup_stale(RegistryHeader * header) if (!snapshot_slot(&slots[i], &snap)) { continue; } + if (snap.pid == 0) { + // pid==0 is never a real entry — only a transient snapshot of a slot + // whose payload is not yet written. Never reclaim it. + continue; + } if (pid_is_alive_or_unreachable(snap.pid)) { continue; } diff --git a/rmw_unix_socket_cpp/src/registry.hpp b/rmw_unix_socket_cpp/src/registry.hpp index 5dab0ce..672683a 100644 --- a/rmw_unix_socket_cpp/src/registry.hpp +++ b/rmw_unix_socket_cpp/src/registry.hpp @@ -27,6 +27,11 @@ enum RegistryEntryType : uint8_t ENTRY_SUBSCRIPTION, ENTRY_SERVICE, ENTRY_CLIENT, + // 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 + // only in the slot's atomic state field, so it costs no struct layout. + ENTRY_RESERVED = 0xFF, }; // User-facing struct, passed to registry_add. Plain data, no atomics. @@ -52,11 +57,12 @@ struct RegistryEntry // seqlock + atomic state machine so readers never need a global mutex. // // Protocol (writer): -// 1. CAS state EMPTY -> target_type (claims the slot atomically) +// 1. CAS state EMPTY -> RESERVED (claims the slot; readers treat as EMPTY) // 2. seq.fetch_add(1, acq_rel) -> seq becomes odd (write in progress) // 3. memcpy payload fields // 4. seq.fetch_add(1, acq_rel) -> seq becomes even (snapshot stable) -// 5. generation.fetch_add(1, acq_rel) -> wakes cached readers +// 5. state.store(target_type, release) -> publishes the committed payload +// 6. generation.fetch_add(1, acq_rel) -> wakes cached readers // // Protocol (reader): // 1. s1 = seq.load(acquire); if (s1 & 1) skip slot (writer in progress)