Data-KDTree-Shared
view release on metacpan or search on metacpan
typedef struct KdHandle {
KdHeader *hdr;
KdReaderSlot *reader_slots; /* KD_READER_SLOTS entries */
uint64_t *occ; /* KD_OCC_WORDS-word slot-occupancy bitmap (trusted layout offset) */
void *base; /* mmap base */
uint64_t nodes_off; /* validated store offsets, cached: never re-read from the peer-writable header */
uint64_t idx_off;
uint64_t node_stride; /* cached */
uint32_t dims; /* cached */
uint32_t capacity; /* cached */
size_t mmap_size;
char *path; /* backing file path (strdup'd) */
int backing_fd; /* memfd or reopened-fd to close on destroy, -1 for file/anon */
uint32_t my_slot_idx; /* UINT32_MAX if all slots taken (no recovery for this handle) */
uint32_t cached_pid; /* getpid() cached at last slot claim */
uint32_t cached_fork_gen; /* kd_fork_gen value at last slot claim */
uint32_t slotless_held; /* rwlock read-locks held with no reader-slot */
int readonly; /* 1 = frozen O_RDONLY/PROT_READ view: lock-free reads, mutation croaks */
} KdHandle;
/* ================================================================
* Futex-based write-preferring read-write lock (reader-slots-only)
* with dead-process recovery
*
* The reader count is NOT stored in a shared counter. It is DISTRIBUTED across
* per-process reader slots: each slot's `rdepth` is that process's entire
* contribution to the lock. A reader publishes its presence in its own slot and
* then re-checks the writer word; a writer publishes the writer word and then
* scans every slot until all live readers' rdepth reach 0. Sequentially-
* consistent store+load on each side (a Dekker handshake) gives mutual exclusion.
*
* Because a reader's whole contribution is ONE atomic word owned by ONE process,
* a crashed reader is recovered by clearing that one slot (CAS its pid to 0) --
* there is no second counter to strand, no orphaned +1, and therefore no
* quiescent force-reset. A reader killed anywhere in rdlock/rdunlock leaves at
* most `rdepth>0` in its dead slot, which the draining writer clears directly, so
* sustained read traffic can never starve a writer. Write-preference is inherent
* in the gate (new readers see wlock!=0 and yield), so there is no reader-count
* yield hack.
* ================================================================ */
#define KD_RWLOCK_SPIN_LIMIT 32
#define KD_LOCK_TIMEOUT_SEC 2 /* FUTEX_WAIT timeout for stale-lock detection / drain re-scan */
static inline void kd_rwlock_spin_pause(void) {
#if defined(__x86_64__) || defined(__i386__)
__asm__ volatile("pause" ::: "memory");
#elif defined(__aarch64__)
__asm__ volatile("yield" ::: "memory");
#else
__asm__ volatile("" ::: "memory");
#endif
}
/* Writer word encoding: WRITER_BIT|pid when write-locked, 0 when free. */
#define KD_RWLOCK_WRITER_BIT 0x80000000U
#define KD_RWLOCK_PID_MASK 0x7FFFFFFFU
#define KD_RWLOCK_WR(pid) (KD_RWLOCK_WRITER_BIT | ((uint32_t)(pid) & KD_RWLOCK_PID_MASK))
/* A zombie (dead but not yet reaped) still answers kill(pid,0) as alive, so a
* process that crashed while holding the lock and lingers unreaped would never
* be recovered. Treat /proc/<pid>/stat state 'Z' as dead. Linux-only (as is
* this module); if /proc is unreadable we fall back to "alive" (safe: we never
* force-recover a possibly-live holder). */
static inline int kd_pid_is_zombie(uint32_t pid) {
char path[32], buf[256];
snprintf(path, sizeof(path), "/proc/%u/stat", (unsigned)pid);
int fd = open(path, O_RDONLY | O_CLOEXEC);
if (fd < 0) return 0;
ssize_t n = read(fd, buf, sizeof(buf) - 1);
close(fd);
if (n <= 0) return 0;
buf[n] = '\0';
/* "pid (comm) state ..."; comm may contain ')', so scan to the last one. */
char *rp = strrchr(buf, ')');
if (!rp || rp + 2 >= buf + n) return 0; /* need ") X" within the bytes read */
return rp[1] == ' ' && rp[2] == 'Z';
}
/* 1 if alive or unknown, 0 if definitely dead. Cannot detect PID reuse: a
* recycled PID reports "alive" and the slot is not reclaimed until that
* process exits. See "Crash Safety" in the POD. */
static inline int kd_pid_alive(uint32_t pid) {
if (pid == 0) return 1; /* no owner recorded, assume alive */
if (kill((pid_t)pid, 0) == -1 && errno == ESRCH) return 0; /* definitely dead */
return !kd_pid_is_zombie(pid); /* kill() also succeeds for a zombie -> treat as dead */
}
/* Force-recover a stale WRITE lock left by a dead writer (held or mid-drain).
* CAS to OUR pid to hold the lock while fixing shared state, then release.
* Using our pid (not a bare WRITER_BIT sentinel) means a subsequent recovering
* process can detect and re-recover if we crash mid-recovery. */
static inline void kd_recover_stale_lock(KdHandle *h, uint32_t observed_wlock) {
KdHeader *hdr = h->hdr;
uint32_t mypid = KD_RWLOCK_WR((uint32_t)getpid());
if (!__atomic_compare_exchange_n(&hdr->wlock, &observed_wlock,
mypid, 0, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED))
return;
/* We now hold the write lock as mypid. No additional shared state needs
* repair here (this module has no seqlock); just release the lock. */
__atomic_store_n(&hdr->wlock, 0, __ATOMIC_RELEASE);
if (__atomic_load_n(&hdr->rwait, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->wlock, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
static const struct timespec kd_lock_timeout = { KD_LOCK_TIMEOUT_SEC, 0 };
/* Process-global fork-generation counter. Incremented in the pthread_atfork
* child callback so every open handle detects a fork transition on the next
* lock call without paying a getpid() syscall on the hot path. */
static uint32_t kd_fork_gen = 1;
static pthread_once_t kd_atfork_once = PTHREAD_ONCE_INIT;
static void kd_on_fork_child(void) {
__atomic_add_fetch(&kd_fork_gen, 1, __ATOMIC_RELAXED);
}
static void kd_atfork_init(void) {
pthread_atfork(NULL, NULL, kd_on_fork_child);
}
/* Occupancy bitmap: set a slot's bit when it is claimed, clear it on clean
* release. SEQ_CST so a set bit is ordered before the slot's rdepth can go
* non-zero (bit set in claim, which precedes any rdlock), letting a writer's
* SEQ_CST bitmap scan never miss a slot a committed reader holds. */
static inline void kd_occ_set(KdHandle *h, uint32_t s) {
__atomic_fetch_or(&h->occ[s >> 6], (uint64_t)1 << (s & 63), __ATOMIC_SEQ_CST);
}
static inline void kd_occ_clear(KdHandle *h, uint32_t s) {
__atomic_fetch_and(&h->occ[s >> 6], ~((uint64_t)1 << (s & 63)), __ATOMIC_SEQ_CST);
}
/* Ensure this process owns a reader slot. Called from the lock helpers so that
* fork()'d children pick up their own slot lazily instead of sharing the
* parent's. Hot-path is a single relaxed load + compare; only on a
* fork-generation mismatch do we touch getpid() and scan slots. */
static inline void kd_claim_reader_slot(KdHandle *h) {
uint32_t cur_gen = __atomic_load_n(&kd_fork_gen, __ATOMIC_RELAXED);
if (__builtin_expect(cur_gen == h->cached_fork_gen && h->my_slot_idx != UINT32_MAX, 1))
return;
/* Cold path -- register the atfork hook once per process, then claim. */
pthread_once(&kd_atfork_once, kd_atfork_init);
/* Re-read after pthread_once: kd_on_fork_child may have bumped it. */
cur_gen = __atomic_load_n(&kd_fork_gen, __ATOMIC_RELAXED);
uint32_t now_pid = (uint32_t)getpid();
h->cached_pid = now_pid;
if (cur_gen != h->cached_fork_gen) h->slotless_held = 0; /* fork: child holds none of the parent's slotless read locks */
h->cached_fork_gen = cur_gen;
h->my_slot_idx = UINT32_MAX;
uint32_t start = now_pid % KD_READER_SLOTS;
/* Pass 1: take a free slot. */
for (uint32_t i = 0; i < KD_READER_SLOTS; i++) {
uint32_t s = (start + i) % KD_READER_SLOTS;
uint32_t expected = 0;
if (__atomic_compare_exchange_n(&h->reader_slots[s].pid,
&expected, now_pid, 0,
__ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
/* Fresh owner holds no read locks yet; clear any stale rdepth left by
* a dead predecessor (its contribution is dropped as we take over). */
__atomic_store_n(&h->reader_slots[s].rdepth, 0, __ATOMIC_RELAXED);
kd_occ_set(h, s); /* mark occupied BEFORE any rdlock can bump rdepth */
h->my_slot_idx = s;
return;
}
}
/* Pass 2: no free slot -- reclaim one whose owner is dead. Safe to take even
* if its rdepth>0: clearing pid drops the dead reader's entire contribution
* (a writer scan ignores rdepth when pid==0) and we reset rdepth to 0 as we
* claim it. */
for (uint32_t i = 0; i < KD_READER_SLOTS; i++) {
uint32_t dpid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (dpid == 0 || dpid == now_pid || kd_pid_alive(dpid)) continue;
uint32_t expected = dpid;
if (__atomic_compare_exchange_n(&h->reader_slots[i].pid, &expected, now_pid, 0,
__ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
__atomic_store_n(&h->reader_slots[i].rdepth, 0, __ATOMIC_RELAXED);
kd_occ_set(h, i);
h->my_slot_idx = i;
return;
}
}
/* Table full -- leave my_slot_idx = UINT32_MAX so this handle takes the
* slotless path (lock still works; recovery of THIS reader's death is the
* documented slotless limitation). */
}
/* Inspect the writer word after a futex-wait timeout. If a dead writer holds
* it, force-recover. Dead READERS need no action here: only a writer that owns
* wlock drains readers, and it clears dead readers inline in its own scan. */
static inline void kd_recover_after_timeout(KdHandle *h) {
uint32_t val = __atomic_load_n(&h->hdr->wlock, __ATOMIC_RELAXED);
if (val >= KD_RWLOCK_WRITER_BIT) {
uint32_t pid = val & KD_RWLOCK_PID_MASK;
if (!kd_pid_alive(pid))
kd_recover_stale_lock(h, val);
}
}
/* Bump/drop the parked-waiter hint. Both readers (blocked at the gate) and
* writers (blocked acquiring wlock) wait on the wlock futex and use this, so
* wrunlock/recover know whether a FUTEX_WAKE is worth a syscall. A waiter
* SIGKILLed while parked leaves rwait over-counted -> at most a spurious wake
* (harmless); it can never under-count, so no wakeup is lost. */
static inline void kd_park(KdHandle *h) {
__atomic_add_fetch(&h->hdr->rwait, 1, __ATOMIC_RELAXED);
}
static inline void kd_unpark(KdHandle *h) {
__atomic_sub_fetch(&h->hdr->rwait, 1, __ATOMIC_RELAXED);
}
/* Publish (inc) / retract (dec) this reader's presence -- its ENTIRE
* contribution to the lock. A slotted reader uses its slot's rdepth; a reader
* that could not claim a slot uses the global slotless_rdepth. inc() is SEQ_CST
* so the wlock re-check that follows it in rdlock forms a Dekker handshake with
* the writer's SEQ_CST wlock-store + rdepth-scan. leave() peels slotless first
* so a slot claimed mid-hold cannot misattribute the decrement. */
static inline void kd_rdepth_inc(KdHandle *h) {
if (h->my_slot_idx != UINT32_MAX) {
__atomic_add_fetch(&h->reader_slots[h->my_slot_idx].rdepth, 1, __ATOMIC_SEQ_CST);
} else {
__atomic_add_fetch(&h->hdr->slotless_rdepth, 1, __ATOMIC_SEQ_CST);
h->slotless_held++;
}
}
static inline void kd_rdepth_dec(KdHandle *h) {
if (h->slotless_held > 0) {
h->slotless_held--;
__atomic_sub_fetch(&h->hdr->slotless_rdepth, 1, __ATOMIC_RELEASE);
} else if (h->my_slot_idx != UINT32_MAX) {
__atomic_sub_fetch(&h->reader_slots[h->my_slot_idx].rdepth, 1, __ATOMIC_RELEASE);
}
}
/* Wake a writer that may be draining readers (it waits on drain_seq). Called
* after every rdepth decrement so a released read lock lets the writer re-scan
* promptly instead of waiting out its timeout. */
static inline void kd_reader_wake_drain(KdHandle *h) {
if (__atomic_load_n(&h->hdr->wlock, __ATOMIC_ACQUIRE) != 0) {
__atomic_add_fetch(&h->hdr->drain_seq, 1, __ATOMIC_RELEASE);
syscall(SYS_futex, &h->hdr->drain_seq, FUTEX_WAKE, 1, NULL, NULL, 0);
}
}
static inline void kd_rwlock_rdlock(KdHandle *h) {
kd_claim_reader_slot(h);
KdHeader *hdr = h->hdr;
for (int spin = 0; ; spin++) {
uint32_t cur = __atomic_load_n(&hdr->wlock, __ATOMIC_ACQUIRE);
if (cur == 0) {
/* Optimistically take the read: publish rdepth, then re-check wlock.
* SEQ_CST inc + SEQ_CST load vs the writer's SEQ_CST wlock CAS +
* SEQ_CST rdepth scan: by the single total order of SEQ_CST ops the
* two sides cannot both miss each other, so we never hold
* concurrently with a writer. */
kd_rdepth_inc(h);
if (__atomic_load_n(&hdr->wlock, __ATOMIC_SEQ_CST) == 0)
return; /* no writer after our publish -> we hold the read lock */
/* A writer appeared during our publish -- yield to it (write-preferring). */
kd_rdepth_dec(h);
kd_reader_wake_drain(h); /* let the draining writer see rdepth drop */
spin = 0;
continue;
}
/* wlock != 0: a writer holds or is acquiring. Recover if it is dead. */
if (cur >= KD_RWLOCK_WRITER_BIT &&
!kd_pid_alive(cur & KD_RWLOCK_PID_MASK)) {
kd_recover_stale_lock(h, cur);
spin = 0;
continue;
}
if (__builtin_expect(spin < KD_RWLOCK_SPIN_LIMIT, 1)) {
kd_rwlock_spin_pause();
continue;
}
kd_park(h);
cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (cur != 0) {
long rc = syscall(SYS_futex, &hdr->wlock, FUTEX_WAIT, cur,
&kd_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
kd_unpark(h);
kd_recover_after_timeout(h);
spin = 0;
continue;
}
}
kd_unpark(h);
spin = 0;
}
}
static inline void kd_rwlock_rdunlock(KdHandle *h) {
kd_rdepth_dec(h); /* RELEASE: drop our entire contribution */
kd_reader_wake_drain(h); /* if a writer is draining, wake it to re-scan */
}
static inline void kd_rwlock_wrlock(KdHandle *h) {
kd_claim_reader_slot(h); /* refresh cached_pid across fork */
KdHeader *hdr = h->hdr;
/* Encode PID in the wlock word itself (0x80000000 | pid) to eliminate any
* crash window between acquiring the lock and storing the owner. */
uint32_t mypid = KD_RWLOCK_WR(h->cached_pid);
/* Phase 1: acquire the writer word (mutual exclusion among writers). */
for (int spin = 0; ; spin++) {
uint32_t expected = 0;
if (__atomic_compare_exchange_n(&hdr->wlock, &expected, mypid,
0, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED))
break;
/* Contended: expected now holds the current wlock value. */
if (expected >= KD_RWLOCK_WRITER_BIT &&
!kd_pid_alive(expected & KD_RWLOCK_PID_MASK)) {
kd_recover_stale_lock(h, expected);
spin = 0;
continue;
}
if (__builtin_expect(spin < KD_RWLOCK_SPIN_LIMIT, 1)) {
kd_rwlock_spin_pause();
continue;
}
kd_park(h);
uint32_t cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (cur != 0) {
long rc = syscall(SYS_futex, &hdr->wlock, FUTEX_WAIT, cur,
&kd_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
kd_unpark(h);
kd_recover_after_timeout(h);
spin = 0;
continue;
}
}
kd_unpark(h);
spin = 0;
}
/* Phase 2: we own wlock, so no NEW reader can join (they see wlock!=0 and
* yield). Drain the readers that were already holding when we won the CAS.
* The SEQ_CST CAS above + the SEQ_CST rdepth loads below are the writer side
* of the Dekker handshake. */
for (;;) {
uint32_t v = __atomic_load_n(&hdr->drain_seq, __ATOMIC_RELAXED); /* snapshot BEFORE scan */
int busy = 0;
/* Visit only OCCUPIED slots via the occupancy bitmap (SEQ_CST: a committed
* reader's bit -- set in claim, before its rdepth++ -- is ordered before
* this scan, so no held slot is skipped). O(KD_OCC_WORDS + live readers)
* instead of O(KD_READER_SLOTS). */
for (uint32_t w = 0; w < KD_OCC_WORDS; w++) {
uint64_t word = __atomic_load_n(&h->occ[w], __ATOMIC_SEQ_CST);
while (word) {
uint32_t i = (w << 6) + (uint32_t)__builtin_ctzll(word);
word &= word - 1; /* consume this bit (local copy) */
uint32_t rd = __atomic_load_n(&h->reader_slots[i].rdepth, __ATOMIC_SEQ_CST);
if (rd == 0) continue; /* occupied but not read-locking now */
uint32_t pid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (pid == 0) continue; /* stale rdepth on a freed slot */
if (!kd_pid_alive(pid)) {
/* Dead reader: drop its pid so the slot no longer counts. Leave
* the occ bit set (harmless -- a later scan hits pid==0 and skips,
* a re-claim re-sets it) to avoid racing a concurrent claimant. */
uint32_t ep = pid;
__atomic_compare_exchange_n(&h->reader_slots[i].pid, &ep, 0,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED);
continue;
}
busy = 1; /* live reader still holding */
}
}
/* A live slotless reader keeps us waiting; a crashed slotless reader that
* cannot be attributed to a pid is the documented slotless limitation. */
if (__atomic_load_n(&hdr->slotless_rdepth, __ATOMIC_SEQ_CST) != 0)
busy = 1;
if (!busy)
return; /* exclusive: wlock held + every rdepth 0 */
/* Wait for a reader to release (drain_seq bump) or time out to re-scan
* (which reclaims any newly-dead slotted reader). */
syscall(SYS_futex, &hdr->drain_seq, FUTEX_WAIT, v, &kd_lock_timeout, NULL, 0);
}
}
static inline void kd_rwlock_wrunlock(KdHandle *h) {
KdHeader *hdr = h->hdr;
__atomic_store_n(&hdr->wlock, 0, __ATOMIC_RELEASE);
if (__atomic_load_n(&hdr->rwait, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->wlock, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
/* ================================================================
* Layout math + create / open / destroy
*
* Layout: Header -> reader_slots[1024] -> occ bitmap -> nodes[capacity] -> build_idx[capacity]
* ================================================================ */
/* Single source of truth for the mmap region layout offsets.
* Layout: Header -> reader_slots[1024] -> occ bitmap -> nodes[capacity] -> build_idx[capacity] */
typedef struct { uint64_t reader_slots, occ, nodes, idx, total; } KdLayout;
static inline uint64_t kd_stride(uint32_t dims) {
return (uint64_t)dims * sizeof(double) + 2 * sizeof(uint32_t) + sizeof(uint64_t); /* coords + left/right + payload */
}
static inline KdLayout kd_layout_for(uint32_t dims, uint32_t capacity) {
KdLayout L;
uint64_t stride = kd_stride(dims);
L.reader_slots = sizeof(KdHeader);
L.occ = L.reader_slots + (uint64_t)KD_READER_SLOTS * sizeof(KdReaderSlot);
L.nodes = L.occ + KD_OCC_BYTES;
L.nodes = (L.nodes + 7) & ~(uint64_t)7;
L.idx = L.nodes + (uint64_t)capacity * stride;
L.idx = (L.idx + 7) & ~(uint64_t)7;
L.total = L.idx + (uint64_t)capacity * sizeof(uint32_t);
return L;
}
static inline uint64_t kd_total_size(uint32_t dims, uint32_t capacity) {
return kd_layout_for(dims, capacity).total;
}
( run in 3.384 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )