Data-HashMap-Shared
view release on metacpan or search on metacpan
shm_generic.h view on Meta::CPAN
/* Per-process slot for dead-process recovery. In the reader-slots-only rwlock a
* reader's ENTIRE contribution to the shared lock is `rdepth` in its OWN slot --
* there is no separate shared reader counter to fall out of sync with it -- so a
* dead reader's contribution is exactly this one word, which a draining writer
* neutralises by clearing the slot's pid (the scan then ignores the slot). No
* orphaned counter can exist, so there is no quiescent force-reset and sustained
* readers cannot starve a writer. _rsv1/_rsv2 are kept only to preserve the
* 16-byte slot size across the already-released builds. */
typedef struct {
uint32_t pid; /* 0 = unclaimed */
uint32_t rdepth; /* read-locks THIS process currently holds (recursion-safe) */
uint32_t _rsv1; /* reserved (was waiters_parked); unused, kept for layout size */
uint32_t _rsv2; /* reserved (was writers_parked); unused, kept for layout size */
} ShmReaderSlot;
/* ---- Process-local handle ---- */
typedef struct ShmHandle_s {
ShmHeader *hdr;
void *nodes;
uint8_t *states;
char *arena;
uint32_t *lru_prev; /* NULL if LRU disabled */
uint32_t *lru_next; /* NULL if LRU disabled */
uint8_t *lru_accessed; /* NULL if LRU disabled -- clock second-chance bit */
uint32_t *expires_at; /* NULL if TTL disabled */
ShmReaderSlot *reader_slots; /* SHM_READER_SLOTS entries */
uint64_t *occ; /* SHM_OCC_WORDS-word slot-occupancy bitmap (trusted layout offset) */
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; /* shm_fork_gen value at last slot claim -- mismatch triggers reclaim */
uint32_t slotless_held; /* rwlock read-locks held with no reader-slot */
uint32_t lock_depth; /* locks this process holds via RDLOCK_GUARD/WRSEQ_GUARD */
uint8_t pending_close; /* DESTROY arrived while lock_depth > 0; free at depth 0 */
int readonly; /* 1 = frozen O_RDONLY/PROT_READ view: reads lock-free, mutation croaks.
A read-only handle NEVER writes the mapping (no rdepth, no clock bit). */
size_t mmap_size;
uint32_t max_mask; /* max_table_cap - 1, for seqlock bounds clamping */
uint32_t iter_pos;
char *copy_buf;
uint32_t copy_buf_size;
uint32_t iterating; /* active iterator count (each + cursors) */
uint32_t iter_gen; /* table_gen snapshot for each() */
uint8_t iter_active; /* 1 = built-in each is in progress */
uint8_t deferred; /* shrink/compact deferred while iterating */
char *path; /* backing file path (strdup'd) */
int backing_fd; /* memfd fd to close on destroy, -1 otherwise */
/* Sharding: if shard_handles != NULL, this is a sharded map dispatcher */
struct ShmHandle_s **shard_handles; /* NULL for single map */
uint32_t num_shards;
uint32_t shard_mask; /* num_shards - 1 (power of 2) */
uint32_t shard_iter; /* current shard for each()/cursor iteration */
} ShmHandle;
/* ---- Cursor (independent iterator) ---- */
typedef struct {
ShmHandle *handle; /* for single maps, direct handle; for sharded, the dispatcher */
ShmHandle *current; /* current shard handle (== handle for single maps) */
SV *owner; /* ref to the map's referent SV; keeps the mmap/handle alive while the cursor lives */
uint32_t iter_pos;
uint32_t gen; /* table_gen snapshot -- reset on mismatch */
uint32_t shard_idx; /* current shard index (0 for single maps) */
uint32_t shard_count; /* total shards (1 for single maps) */
char *copy_buf;
uint32_t copy_buf_size;
} ShmCursor;
/* Grow a copy buffer to hold `needed` bytes; returns 0 on OOM */
static inline int shm_grow_buf(char **buf, uint32_t *cap, uint32_t needed) {
if (needed == 0) needed = 1;
if (needed <= *cap) return 1;
uint32_t ns = *cap ? *cap : 64;
while (ns < needed) {
uint32_t next = ns * 2;
if (next <= ns) { ns = needed; break; } /* overflow guard */
ns = next;
}
char *nb = (char *)realloc(*buf, ns);
if (!nb) return 0;
*buf = nb;
*cap = ns;
return 1;
}
static inline int shm_ensure_copy_buf(ShmHandle *h, uint32_t needed) {
return shm_grow_buf(&h->copy_buf, &h->copy_buf_size, needed);
}
static inline int shm_cursor_ensure_copy_buf(ShmCursor *c, uint32_t needed) {
return shm_grow_buf(&c->copy_buf, &c->copy_buf_size, needed);
}
/* ---- Hash functions (xxHash, XXH3) ---- */
static inline uint64_t shm_hash_int64(int64_t key) {
return XXH3_64bits(&key, sizeof(key));
}
static inline uint64_t shm_hash_string(const char *data, uint32_t len) {
return XXH3_64bits(data, (size_t)len);
}
/* ---- Futex-based read-write lock ---- */
#define SHM_RWLOCK_SPIN_LIMIT 32
#define SHM_LOCK_TIMEOUT_SEC 2 /* FUTEX_WAIT timeout for stale lock detection */
static inline void shm_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 SHM_RWLOCK_WRITER_BIT 0x80000000U
#define SHM_RWLOCK_PID_MASK 0x7FFFFFFFU
#define SHM_RWLOCK_WR(pid) (SHM_RWLOCK_WRITER_BIT | ((uint32_t)(pid) & SHM_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 shm_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 shm_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 !shm_pid_is_zombie(pid); /* kill() also succeeds for a zombie -> treat as dead */
}
/* Forward declaration -- defined later in the LRU helpers section. */
static void shm_lru_rebuild_if_corrupt(ShmHandle *h);
/* Force-recover a stale write lock left by a dead process.
* CAS to OUR pid to hold the lock while fixing seqlock, 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 shm_recover_stale_lock(ShmHandle *h, uint32_t observed_wlock) {
ShmHeader *hdr = h->hdr;
uint32_t mypid = SHM_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. Repair shared state -- the
* seqlock counter (if dead writer left it odd) and the LRU doubly-
* linked list (if dead writer left it one-way-broken) -- while no
* other process can mutate them. */
uint32_t seq = __atomic_load_n(&hdr->seq, __ATOMIC_RELAXED);
if (seq & 1)
__atomic_store_n(&hdr->seq, seq + 1, __ATOMIC_RELEASE);
shm_lru_rebuild_if_corrupt(h);
__atomic_add_fetch(&hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
/* 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 shm_lock_timeout = { SHM_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 shm_fork_gen = 1;
static pthread_once_t shm_atfork_once = PTHREAD_ONCE_INIT;
static void shm_on_fork_child(void) {
__atomic_add_fetch(&shm_fork_gen, 1, __ATOMIC_RELAXED);
}
static void shm_atfork_init(void) {
pthread_atfork(NULL, NULL, shm_on_fork_child);
}
/* 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. */
/* 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 shm_occ_set(ShmHandle *h, uint32_t s) {
__atomic_fetch_or(&h->occ[s >> 6], (uint64_t)1 << (s & 63), __ATOMIC_SEQ_CST);
}
static inline void shm_occ_clear(ShmHandle *h, uint32_t s) {
__atomic_fetch_and(&h->occ[s >> 6], ~((uint64_t)1 << (s & 63)), __ATOMIC_SEQ_CST);
}
static inline void shm_claim_reader_slot(ShmHandle *h) {
uint32_t cur_gen = __atomic_load_n(&shm_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(&shm_atfork_once, shm_atfork_init);
/* Re-read after pthread_once: shm_on_fork_child may have bumped it. */
cur_gen = __atomic_load_n(&shm_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 % SHM_READER_SLOTS;
/* Pass 1: take a free slot. */
for (uint32_t i = 0; i < SHM_READER_SLOTS; i++) {
uint32_t s = (start + i) % SHM_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);
shm_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 < SHM_READER_SLOTS; i++) {
uint32_t dpid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (dpid == 0 || dpid == now_pid || shm_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);
shm_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 the lock (which also rebuilds the LRU list if it was left
* half-linked, all under the recovered write lock). 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 shm_recover_after_timeout(ShmHandle *h) {
ShmHeader *hdr = h->hdr;
uint32_t val = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (val >= SHM_RWLOCK_WRITER_BIT) {
uint32_t pid = val & SHM_RWLOCK_PID_MASK;
if (!shm_pid_alive(pid))
shm_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 shm_park(ShmHandle *h) {
__atomic_add_fetch(&h->hdr->rwait, 1, __ATOMIC_RELAXED);
}
static inline void shm_unpark(ShmHandle *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. dec() peels slotless first so
* a slot claimed mid-hold cannot misattribute the decrement. */
static inline void shm_rdepth_inc(ShmHandle *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 shm_rdepth_dec(ShmHandle *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 shm_reader_wake_drain(ShmHandle *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 shm_rwlock_rdlock(ShmHandle *h) {
/* Frozen (read-only) view: the file is sealed and immutable, so no writer can
* ever exist (mutators croak, a read-write reopen of a sealed file is refused).
* There is nothing to exclude -- and the mapping is PROT_READ, so publishing
* rdepth into a reader slot would fault. Skip the lock entirely. */
if (h->readonly) return;
shm_claim_reader_slot(h);
ShmHeader *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. */
shm_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). */
shm_rdepth_dec(h);
shm_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 >= SHM_RWLOCK_WRITER_BIT &&
!shm_pid_alive(cur & SHM_RWLOCK_PID_MASK)) {
shm_recover_stale_lock(h, cur);
spin = 0;
continue;
}
if (__builtin_expect(spin < SHM_RWLOCK_SPIN_LIMIT, 1)) {
shm_rwlock_spin_pause();
continue;
}
shm_park(h);
cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (cur != 0) {
long rc = syscall(SYS_futex, &hdr->wlock, FUTEX_WAIT, cur,
&shm_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
shm_unpark(h);
shm_recover_after_timeout(h);
spin = 0;
continue;
}
}
shm_unpark(h);
spin = 0;
}
}
static inline void shm_rwlock_rdunlock(ShmHandle *h) {
if (h->readonly) return; /* frozen view took no lock -- see shm_rwlock_rdlock */
shm_rdepth_dec(h); /* RELEASE: drop our entire contribution */
shm_reader_wake_drain(h); /* if a writer is draining, wake it to re-scan */
}
static inline void shm_rwlock_wrlock(ShmHandle *h) {
/* A frozen (read-only) handle never reaches here: every mutator XSUB croaks
* on h->readonly BEFORE taking any lock (including the counter fast paths
* that write under the READ lock). This function stays pure C with no Perl
* API -- xt/slotless_reader_recovery.t compiles it standalone (plain cc, no
* perl.h), so a Perl_croak here would break that harness. */
shm_claim_reader_slot(h); /* refresh cached_pid across fork */
ShmHeader *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 = SHM_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 >= SHM_RWLOCK_WRITER_BIT &&
!shm_pid_alive(expected & SHM_RWLOCK_PID_MASK)) {
shm_recover_stale_lock(h, expected);
spin = 0;
continue;
}
if (__builtin_expect(spin < SHM_RWLOCK_SPIN_LIMIT, 1)) {
shm_rwlock_spin_pause();
continue;
}
shm_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,
&shm_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
shm_unpark(h);
shm_recover_after_timeout(h);
spin = 0;
continue;
}
}
shm_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(SHM_OCC_WORDS + live readers)
* instead of O(SHM_READER_SLOTS). */
for (uint32_t w = 0; w < SHM_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 (!shm_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, &shm_lock_timeout, NULL, 0);
}
}
static inline void shm_rwlock_wrunlock(ShmHandle *h) {
ShmHeader *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);
}
/* ---- Seqlock (lock-free readers) ---- */
static inline uint32_t shm_seqlock_read_begin(ShmHandle *h) {
ShmHeader *hdr = h->hdr;
int spin = 0;
for (;;) {
uint32_t s = __atomic_load_n(&hdr->seq, __ATOMIC_ACQUIRE);
if (__builtin_expect((s & 1) == 0, 1)) return s;
if (__builtin_expect(spin < 100000, 1)) {
shm_rwlock_spin_pause();
spin++;
continue;
}
/* Prolonged odd seq -- check for dead writer */
uint32_t val = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (val >= SHM_RWLOCK_WRITER_BIT) {
uint32_t pid = val & SHM_RWLOCK_PID_MASK;
if (!shm_pid_alive(pid)) {
shm_recover_stale_lock(h, val);
spin = 0;
continue;
}
}
/* Writer is alive, yield CPU */
struct timespec ts = {0, 1000000}; /* 1ms */
nanosleep(&ts, NULL);
spin = 0;
}
}
static inline int shm_seqlock_read_retry(uint32_t *seq, uint32_t start) {
__atomic_thread_fence(__ATOMIC_ACQUIRE); /* ensure data loads complete before retry check */
return __atomic_load_n(seq, __ATOMIC_RELAXED) != start;
}
static inline void shm_seqlock_write_begin(uint32_t *seq) {
__atomic_add_fetch(seq, 1, __ATOMIC_RELEASE); /* seq becomes odd */
/* StoreStore (Linux write_seqcount_begin's smp_wmb): the odd seq must be
* visible before the entry writes that follow, or an ARM64 reader could
* load an even seq yet observe half-written data and pass read_retry. */
__atomic_thread_fence(__ATOMIC_RELEASE);
}
static inline void shm_seqlock_write_end(uint32_t *seq) {
__atomic_add_fetch(seq, 1, __ATOMIC_RELEASE); /* seq becomes even */
}
/* ---- Arena allocator ---- */
static inline uint32_t shm_next_pow2(uint32_t v);
static inline uint32_t shm_arena_round_up(uint32_t len) {
if (len < SHM_ARENA_MIN_ALLOC) return SHM_ARENA_MIN_ALLOC;
return shm_next_pow2(len);
}
static inline int shm_arena_class_index(uint32_t alloc_size) {
if (alloc_size <= SHM_ARENA_MIN_ALLOC) return 0;
if (alloc_size > (SHM_ARENA_MIN_ALLOC << (SHM_ARENA_NUM_CLASSES - 1))) return -1;
return 32 - __builtin_clz(alloc_size - 1) - 4; /* log2(alloc_size) - 4 */
}
static inline uint32_t shm_arena_alloc(ShmHeader *hdr, char *arena, uint32_t len) {
uint32_t asize = shm_arena_round_up(len);
int cls = shm_arena_class_index(asize);
if (cls >= 0 && hdr->arena_free[cls] != 0) {
uint32_t head = hdr->arena_free[cls];
/* Free-list heads are peer-writable: a wild head would send both the
* pop read below and the caller's string store out of bounds. Gate
* against arena_cap (read-path clamp discipline); on corruption treat
* the class as empty and fall back to bump allocation. */
if ((uint64_t)head + asize <= hdr->arena_cap) {
uint32_t next;
memcpy(&next, arena + head, sizeof(uint32_t));
hdr->arena_free[cls] = next;
return head;
}
}
if (cls < 0) {
/* Large request: first-fit over the large free list before bumping. */
uint32_t prev = 0, cur = hdr->arena_large_free;
while (cur != 0) {
uint32_t next, blk;
shm_generic.h view on Meta::CPAN
c->current = h;
}
c->gen = c->current->hdr->table_gen;
c->current->iterating++;
return c;
}
static inline void shm_cursor_destroy(ShmCursor *c) {
if (!c) return;
ShmHandle *cur = c->current;
if (cur && cur->iterating > 0)
cur->iterating--;
free(c->copy_buf);
free(c);
}
/* ================================================================
* v9 -> v10 on-disk migration (offline structural upcast). The v10
* layout inserts a 128-byte reader-slot occupancy bitmap between the
* reader-slot table and the arena; a v9 file is byte-identical apart
* from that missing region, so migration is a pure re-header + shift --
* no re-hashing and no old library required. Returns 1 if the file was
* upgraded, 0 if it is already current (v10), -1 on error (errbuf set).
* The caller MUST ensure no process has the file mapped.
* ================================================================ */
#define SHMUP_ERR(...) do { if (errbuf) snprintf(errbuf, SHM_ERR_BUFLEN, __VA_ARGS__); } while (0)
#define SHM_UPGRADE_SRC_VERSION 9U /* the single previous on-disk format this transform upgrades from */
/* Tripwire: this function implements EXACTLY the v9 -> v10 (occupancy-bitmap
* insertion) transform and stamps the result as SHM_VERSION. If the on-disk
* format is ever bumped again, compilation fails here until the new step is
* added and this assertion (and SHM_UPGRADE_SRC_VERSION) are updated -- so the
* tool can never silently apply the wrong transform and mis-stamp a file. */
SHM_STATIC_ASSERT(SHM_VERSION == 10U,
"shm_upgrade_file only knows v9 -> v10; extend it when SHM_VERSION changes");
static int shm_upgrade_file(const char *path, char *errbuf) {
if (errbuf) errbuf[0] = '\0';
int fd = open(path, O_RDWR | O_NOFOLLOW | O_CLOEXEC);
if (fd < 0) { SHMUP_ERR("open %s: %s", path, strerror(errno)); return -1; }
struct stat st;
if (fstat(fd, &st) != 0) { SHMUP_ERR("fstat %s: %s", path, strerror(errno)); close(fd); return -1; }
if ((uint64_t)st.st_size < sizeof(ShmHeader)) {
SHMUP_ERR("%s: too small to be a HashMap file (%lld bytes)", path, (long long)st.st_size);
close(fd); return -1;
}
ShmHeader hdr;
if (pread(fd, &hdr, sizeof hdr, 0) != (ssize_t)sizeof hdr) {
SHMUP_ERR("read header %s: %s", path, strerror(errno)); close(fd); return -1;
}
if (hdr.magic != SHM_MAGIC) { SHMUP_ERR("%s: bad magic (not a HashMap::Shared file)", path); close(fd); return -1; }
if (hdr.version == SHM_VERSION) { close(fd); return 0; } /* already current */
if (hdr.version != SHM_UPGRADE_SRC_VERSION) {
SHMUP_ERR("%s: unsupported source version %u (migrates %u -> %u only)",
path, hdr.version, SHM_UPGRADE_SRC_VERSION, (unsigned)SHM_VERSION);
close(fd); return -1;
}
if (hdr.wlock & 0x80000000U) {
/* A crashed writer leaves its pid here forever, so do not assert the
* holder is live without asking: reporting a dead pid as "a live
* writer" sends the operator hunting a process that does not exist. */
uint32_t wpid = hdr.wlock & 0x7FFFFFFFU;
if (shm_pid_alive(wpid))
SHMUP_ERR("%s: locked by a live writer (pid %u); ensure no process is using it",
path, wpid);
else
SHMUP_ERR("%s: holds a stale write lock from pid %u, which is gone. "
"If no process is using this file, the lock is a crash "
"residue -- open it once with the library to recover it, "
"then re-run the upgrade", path, wpid);
close(fd); return -1;
}
if ((uint64_t)st.st_size != hdr.total_size) {
SHMUP_ERR("%s: size mismatch (file=%lld, header=%llu)",
path, (long long)st.st_size, (unsigned long long)hdr.total_size);
close(fd); return -1;
}
uint64_t rss = (uint64_t)SHM_READER_SLOTS * sizeof(ShmReaderSlot);
uint64_t occ_off = hdr.reader_slots_off + rss; /* v9 arena start == new occ-region start */
if (hdr.reader_slots_off < sizeof(ShmHeader) || occ_off > hdr.total_size) {
SHMUP_ERR("%s: reader-slot region out of range (corrupt header)", path); close(fd); return -1;
}
int has_arena = (hdr.arena_off != 0);
if (has_arena ? (hdr.arena_off != occ_off) : (hdr.total_size != occ_off)) {
SHMUP_ERR("%s: unexpected v9 layout; refusing to migrate", path); close(fd); return -1;
}
uint64_t old_size = hdr.total_size, new_size = old_size + SHM_OCC_BYTES;
uint8_t *old = (uint8_t *)malloc((size_t)old_size);
uint8_t *neu = (uint8_t *)malloc((size_t)new_size);
if (!old || !neu) { SHMUP_ERR("out of memory"); free(old); free(neu); close(fd); return -1; }
if (pread(fd, old, (size_t)old_size, 0) != (ssize_t)old_size) {
SHMUP_ERR("read %s: %s", path, strerror(errno)); free(old); free(neu); close(fd); return -1;
}
close(fd);
/* header..reader_slots copied verbatim; insert a zeroed occ region; shift the arena */
memcpy(neu, old, (size_t)occ_off);
memset(neu + occ_off, 0, (size_t)SHM_OCC_BYTES);
memcpy(neu + occ_off + SHM_OCC_BYTES, old + occ_off, (size_t)(old_size - occ_off));
free(old);
ShmHeader *nh = (ShmHeader *)neu;
nh->version = SHM_VERSION;
nh->total_size = new_size;
if (has_arena) nh->arena_off += SHM_OCC_BYTES;
/* reset transient lock/recovery state so the migrated file opens clean */
nh->wlock = 0; nh->rwait = 0; nh->seq = 0; nh->drain_seq = 0; nh->slotless_rdepth = 0;
memset(neu + hdr.reader_slots_off, 0, (size_t)rss); /* drop any stale reader PIDs */
/* write to a sibling temp file, fsync, atomic rename; preserve mode */
size_t plen = strlen(path);
char *tmp = (char *)malloc(plen + 16);
if (!tmp) { SHMUP_ERR("out of memory"); free(neu); return -1; }
snprintf(tmp, plen + 16, "%s.upgrade-tmp", path);
int tfd = open(tmp, O_RDWR | O_CREAT | O_TRUNC | O_NOFOLLOW | O_CLOEXEC, st.st_mode & 07777);
if (tfd < 0) { SHMUP_ERR("create %s: %s", tmp, strerror(errno)); free(neu); free(tmp); return -1; }
(void)fchmod(tfd, st.st_mode & 07777);
/* rename() installs a NEW inode, so the group of a group-shared map would
* otherwise revert to the caller's primary group and lock out the very
* peers the mode was widened for. Best-effort: an unprivileged caller
* cannot always restore an arbitrary owner, and failing for that reason
* should not fail the upgrade. */
int chown_rc = fchown(tfd, (uid_t)-1, st.st_gid);
(void)chown_rc; /* best-effort; see above */
/* write() may transfer less than asked on a large file; loop. */
int ok = 1;
( run in 2.306 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )