Skip to content

Fix IPC ownership and recovery races #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
leducp merged 1 commit into from
Sep 24, 2026
Merged

Fix IPC ownership and recovery races #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
680 changes: 424 additions & 256 deletions ARCHITECTURE.md
View file
Open in desktop

Large diffs are not rendered by default.

7 changes: 6 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,12 @@ invalid value falls back to `0600` with a warning on stderr.
|----------|-------------|-------|
| Linux | `shm_open` / `mmap` | `SYS_futex` |
| macOS | `shm_open` / `mmap` | `__ulock_wait` / `__ulock_wake` |
| Windows | `CreateFileMapping` / `MapViewOfFile` | `WaitOnAddress` / `WakeByAddressAll` |
| Windows | `CreateFileMapping` / `MapViewOfFile` | `WaitOnAddress` / `WakeByAddressAll` (*) |

(*) **Windows limitation:** `WakeByAddressAll` wakes only the calling process.
A cross-process `receive()` may wait until timeout; unread messages can overflow
the ring during that wait. Use a timeout within the ring's buffering budget,
or poll. See [ARCHITECTURE.md](ARCHITECTURE.md) (Platform Abstraction).

Actively validated on Linux x86-64, Linux ARM64 (Raspberry Pi 4B, 12 h continuous stress), and Darwin ARM64 (Apple Silicon, 12 h continuous stress: 2660 passes, 0 failures, 0 reorders) via `scripts/validate.sh` and `tests/endurance.sh`.

Expand Down
10 changes: 5 additions & 5 deletions benchmarks/microbench.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,16 +37,16 @@ static void BM_TreiberPopPush(benchmark::State& state)
SHM_NAME, kickmsg::channel::PubSub, cfg, "bench");

auto* base = region.base();
auto* hdr = region.header();
auto* header = region.header();

for (auto _ : state)
{
uint32_t idx = kickmsg::treiber_pop(hdr->free_top, base, hdr);
uint32_t idx = kickmsg::treiber_pop(header->free_top, base, region.geometry());
benchmark::DoNotOptimize(idx);
if (idx != kickmsg::INVALID_SLOT)
{
auto* slot = kickmsg::slot_at(base, hdr, idx);
kickmsg::treiber_push(hdr->free_top, slot, idx);
auto* slot = kickmsg::slot_at(base, region.geometry(), idx);
kickmsg::treiber_push(header->free_top, slot, idx);
}
}

Expand Down Expand Up @@ -249,7 +249,7 @@ static void BM_CASAdmission(benchmark::State& state)
auto region = kickmsg::SharedRegion::create(
SHM_NAME, kickmsg::channel::PubSub, cfg, "bench");

auto* ring = kickmsg::sub_ring_at(region.base(), region.header(), 0);
auto* ring = kickmsg::sub_ring_at(region.base(), region.geometry(), 0);
// Set ring to Live so CAS admission succeeds
ring->state_flight.store(
kickmsg::ring::make_packed(kickmsg::ring::Live),
Expand Down
8 changes: 4 additions & 4 deletions examples/hello_diagnose.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,23 +56,23 @@ int main()
std::cout << "\n=== Step 2: Inject faults (simulating publisher crashes) ===\n";

auto* base = region.base();
auto* hdr = region.header();
auto* header = region.header();

// Fault 1: Lock a ring entry (simulates publisher crash mid-commit)
{
auto* ring = kickmsg::sub_ring_at(base, hdr, 0);
auto* ring = kickmsg::sub_ring_at(base, region.geometry(), 0);
auto* entries = kickmsg::ring_entries(ring);
// Pretend a publisher claimed pos=write_pos and locked the entry
uint64_t wp = ring->write_pos.load(std::memory_order_acquire);
ring->write_pos.store(wp + 1, std::memory_order_release);
entries[wp & hdr->sub_ring_mask].sequence.store(
entries[wp & header->sub_ring_mask].sequence.store(
kickmsg::seq_lock(wp), std::memory_order_release);
std::cout << " Injected: stale lock at ring 0, pos " << wp << "\n";
}

// Fault 2: Stuck ring (simulates subscriber teardown timeout after publisher crash)
{
auto* ring = kickmsg::sub_ring_at(base, hdr, 1);
auto* ring = kickmsg::sub_ring_at(base, region.geometry(), 1);
ring->state_flight.store(
kickmsg::ring::make_packed(kickmsg::ring::Free, 1),
std::memory_order_release);
Expand Down
19 changes: 10 additions & 9 deletions examples/hello_zerocopy.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,27 +42,28 @@ int main()
// Publish a few "frames"
for (uint32_t i = 0; i < 3; ++i)
{
auto [ptr, max_size] = pub.allocate();
if (ptr == nullptr)
auto slot = pub.allocate();
if (not slot.valid())
{
std::cerr << "Pool exhausted at frame " << i << "\n";
continue;
}

ImageHeader hdr{640, 480, 3, i};
std::memcpy(ptr, &hdr, sizeof(hdr));
pub.publish(sizeof(hdr));
// Written straight into shared memory: no staging buffer, no copy.
ImageHeader header{640, 480, 3, i};
std::memcpy(slot.data(), &header, sizeof(header));
slot.publish(sizeof(header));

std::cout << "Published frame " << i << " (640x480x3)\n";
}

// Zero-copy receive: view points directly into shared memory
while (auto view = sub.try_receive_view())
{
auto const* hdr = static_cast<ImageHeader const*>(view->data());
std::cout << "Received frame " << hdr->frame_id
<< " (" << hdr->width << "x" << hdr->height
<< "x" << hdr->channels << ")"
auto const* header = static_cast<ImageHeader const*>(view->data());
std::cout << "Received frame " << header->frame_id
<< " (" << header->width << "x" << header->height
<< "x" << header->channels << ")"
<< " — zero-copy, " << view->len() << " bytes pinned\n";

// The slot remains pinned while 'view' is alive.
Expand Down
77 changes: 52 additions & 25 deletions include/kickmsg/Blackboard.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@ namespace kickmsg
{
namespace blackboard
{
constexpr uint32_t VERSION = 1;
/// 2: payload and key bytes are accessed as relaxed atomics; a version-1 peer's
/// plain copies would race them again.
constexpr uint32_t VERSION = 2;
constexpr uint64_t MAGIC = 0x214B4C424B43494BULL; // "KICKBLK!"
constexpr std::size_t KEY_MAX = 128;
constexpr std::size_t NODE_NAME_MAX = 64;
Expand Down Expand Up @@ -86,6 +88,16 @@ namespace kickmsg
std::string owner_node;
bool owner_alive;
};

/// Validated local copy of the board geometry; offsets and bounds use only this.
/// The shared header stays peer-writable after open.
struct Geometry
{
uint32_t capacity{0};
std::size_t max_value_size{0}; ///< Value limit; never the padded stride
std::size_t value_stride{0};
std::size_t values_offset{0}; ///< Byte offset of the first value cell
};
}

/// One value cell. Written between the odd and even `publish` stores and
Expand Down Expand Up @@ -125,7 +137,9 @@ namespace kickmsg
std::atomic<uint64_t> key_hash; ///< resolve pre-filter only, never an identity proof
std::atomic<uint64_t> declared_at_ns;
uint8_t _pad1[8];
char key[blackboard::KEY_MAX]; ///< may be unterminated
/// Key text as atomic words: read unlocked while a claim rewrites it. Access only
/// through bb_store_key() / bb_load_key(). May be unterminated.
std::atomic<uint64_t> key[blackboard::KEY_MAX / sizeof(uint64_t)];
char owner_node[blackboard::NODE_NAME_MAX]; ///< may be unterminated
uint8_t _padding[128];
};
Expand All @@ -135,6 +149,8 @@ namespace kickmsg
"entry stride must keep every entry cache-line aligned");
static_assert(offsetof(BlackboardEntry, key) == 64,
"the guard words must occupy exactly the first cache line");
static_assert(blackboard::KEY_MAX % sizeof(uint64_t) == 0,
"key storage is whole atomic words");
static_assert(offsetof(BlackboardEntry, _padding) == 256,
"BlackboardEntry field offsets must match the expected 256 B prefix");
static_assert(std::is_standard_layout<BlackboardEntry>::value,
Expand Down Expand Up @@ -195,8 +211,15 @@ namespace kickmsg
"BlackboardHeader is placed in shared memory via reinterpret_cast");

BlackboardEntry* bb_entry_at(void* base, uint32_t idx);
BlackboardCell* bb_cell_at(void* base, uint32_t idx, uint64_t parity);
uint8_t* bb_cell_payload(BlackboardCell* cell);
BlackboardCell* bb_cell_at(void* base, blackboard::Geometry const& geometry, uint32_t idx, uint64_t parity);
/// Payload words following the cell. Accessed only as whole relaxed words, so a
/// reader overlapping a writer never mixes access sizes.
std::atomic<uint64_t>* bb_cell_words(BlackboardCell* cell);

/// Store `len` <= KEY_MAX key bytes, zero-padded to KEY_MAX, as relaxed words.
void bb_store_key(BlackboardEntry* entry, char const* key, std::size_t len);
/// Relaxed word copy of the stored key, cut at the first NUL or KEY_MAX.
std::string bb_load_key(BlackboardEntry const* entry);

uint64_t bb_config_hash(blackboard::Config const& cfg);

Expand Down Expand Up @@ -292,17 +315,18 @@ namespace kickmsg

private:
friend class Blackboard;
Writer(void* base, uint32_t entry_idx, uint64_t tenancy,
uint64_t writes, uint64_t owner_pid, std::string key);

void* base_{nullptr};
uint32_t entry_idx_{INVALID_SLOT};
uint64_t tenancy_{0};
uint64_t writes_{0}; ///< sole owner, so this counter lives in the handle
Writer(void* base, blackboard::Geometry const& geometry, uint32_t entry_idx,
uint64_t tenancy, uint64_t writes, uint64_t owner_pid, std::string key);

void* base_{nullptr};
blackboard::Geometry geometry_{};
uint32_t entry_idx_{INVALID_SLOT};
uint64_t tenancy_{0};
uint64_t writes_{0}; ///< sole owner, so this counter lives in the handle
/// Declaring process. A Writer inherited across fork() writes and
/// releases nothing: the claim stays the parent's.
uint64_t owner_pid_{0};
std::string key_;
uint64_t owner_pid_{0};
std::string key_;
};

/// Declared read interest in one key. Copyable: it owns nothing.
Expand Down Expand Up @@ -359,19 +383,20 @@ namespace kickmsg

private:
friend class Blackboard;
Reader(void* base, std::string key);
Reader(void* base, blackboard::Geometry const& geometry, std::string key);

/// Resolves entry_idx_ if it is unset or its tenancy moved.
/// Returns false when no active entry holds this key.
bool resolve() const;

void* base_{nullptr};
void* base_{nullptr};
blackboard::Geometry geometry_{};
/// INVALID_SLOT until the key first materializes: observing a key
/// before its writer exists is a supported use.
mutable uint32_t entry_idx_{INVALID_SLOT};
mutable uint64_t tenancy_{0};
uint64_t key_hash_{0};
std::string key_;
mutable uint32_t entry_idx_{INVALID_SLOT};
mutable uint64_t tenancy_{0};
uint64_t key_hash_{0};
std::string key_;
};

/// Claim exclusive ownership of `key`.
Expand Down Expand Up @@ -481,6 +506,7 @@ namespace kickmsg
std::string const& name() const { return name_; }
uint32_t capacity() const;
std::size_t max_value_size() const;
blackboard::Geometry const& geometry() const { return geometry_; }

BlackboardHeader* header() { return static_cast<BlackboardHeader*>(base_); }
BlackboardHeader const* header() const { return static_cast<BlackboardHeader const*>(base_); }
Expand All @@ -492,7 +518,7 @@ namespace kickmsg
void init_as_creator(blackboard::Config const& cfg);

/// Sweep body, run by a caller that already holds the board lock.
uint32_t sweep_locked(BlackboardHeader* h);
uint32_t sweep_locked();

struct EntryRead
{
Expand All @@ -508,11 +534,12 @@ namespace kickmsg
std::error_code read_entry(uint32_t i, EntryRead& out,
std::vector<uint8_t>* value) const;

SharedMemory shm_;
std::string name_;
std::string owner_name_;
void* base_{nullptr};
std::size_t size_{0};
SharedMemory shm_;
std::string name_;
std::string owner_name_;
void* base_{nullptr};
std::size_t size_{0};
blackboard::Geometry geometry_{};
};
}

Expand Down
Loading
Loading