Data-HashMap-Shared
view release on metacpan or search on metacpan
slot, published before the slot can hold a lock) lets the writer visit
only occupied slots rather than all 1024. Beyond 1024 simultaneous handles
per map, a handle that cannot claim a slot proceeds "slotless"; see
"Reader-slot exhaustion" for the one case that recovery cannot cover.
The same path validates and rebuilds the LRU doubly-linked list if a dead
writer left it inconsistent. "stat_recoveries" in "stats" counts every
recovery event.
Recovery uses "kill($pid, 0)" for liveness, which cannot distinguish a
reused PID from the original. Hitting a false "alive" requires a process
to die in the brief window it holds a read lock and the kernel to cycle
through the entire PID space back to that exact number within the
~2-second recovery window and hand it to a long-lived process -- i.e. a
runaway fork storm. Even then the effect is bounded: writers stall until
the recycled process exits; reads are unaffected and no data is corrupted.
Writer-crash recovery is immune (the writer PID lives in the lock word and
is reclaimed independently of the slot table).
Limitation: PID-based recovery assumes all processes share the same PID
namespace. Cross-container sharing (different PID namespaces) is not
print "permanent=$pv ttl_remaining=$pr\n";
}
print "\nSleeping 3 seconds...\n";
sleep 3;
my $v = shm_si_get $map, "counter";
printf "counter=%s (expired: %s)\n", $v // 'undef', defined $v ? 'no' : 'yes';
my $p = shm_si_get $map, "permanent";
printf "permanent=%s (still alive)\n", $p;
$map->unlink;
lib/Data/HashMap/Shared.pm view on Meta::CPAN
the writer visit only occupied slots rather than all 1024. Beyond 1024
simultaneous handles per map, a handle that cannot claim a slot proceeds
"slotless"; see L</"Reader-slot exhaustion"> for the one case that
recovery cannot cover.
The same path validates and rebuilds the LRU doubly-linked list if a
dead writer left it inconsistent. C<stat_recoveries> in C<stats> counts
every recovery event.
Recovery uses C<kill($pid, 0)> for liveness, which cannot distinguish a
reused PID from the original. Hitting a false "alive" requires a process to
die in the brief window it holds a read lock B<and> the kernel to cycle
through the entire PID space back to that exact number within the ~2-second
recovery window B<and> hand it to a long-lived process -- i.e. a runaway fork
storm. Even then the effect is bounded: writers stall until the recycled
process exits; reads are unaffected and no data is corrupted. Writer-crash
recovery is immune (the writer PID lives in the lock word and is reclaimed
independently of the slot table).
B<Limitation>: PID-based recovery assumes all processes share the same
PID namespace. Cross-container sharing (different PID namespaces) is not
shm_generic.h view on Meta::CPAN
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) {
shm_generic.h view on Meta::CPAN
#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
shm_generic.h view on Meta::CPAN
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
shm_generic.h view on Meta::CPAN
/* 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) {
shm_generic.h view on Meta::CPAN
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);
shm_generic.h view on Meta::CPAN
* 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);
shm_generic.h view on Meta::CPAN
* 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 */
}
shm_generic.h view on Meta::CPAN
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;
}
shm_generic.h view on Meta::CPAN
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) {
xt/edge-cases.t view on Meta::CPAN
}
# unlink: instance method
{
my $path = tmpfile();
my $map = Data::HashMap::Shared::II->new($path, 100);
shm_ii_put $map, 1, 42;
ok(-f $path, 'backing file exists');
ok($map->unlink, 'instance unlink returns true');
ok(!-f $path, 'backing file removed after unlink');
# map still works (mmap stays alive after unlink)
my $v = shm_ii_get $map, 1;
is($v, 42, 'map still readable after unlink');
}
# unlink: class method
{
my $path = tmpfile();
my $map = Data::HashMap::Shared::II->new($path, 100);
undef $map;
ok(-f $path, 'backing file exists before class unlink');
xt/lru-ttl.t view on Meta::CPAN
shm_ii_put $map, 1, 10;
shm_ii_put_ttl $map, 2, 20, 100; # long TTL
sleep 4;
my @k = shm_ii_keys $map;
# key 1 should be expired during iteration, key 2 should survive
# Note: keys iteration may or may not lazily expire
# But get should definitely expire
ok(!defined(shm_ii_get $map, 1), 'key 1 expired');
is(shm_ii_get $map, 2, 20, 'key 2 still alive');
unlink $path;
}
# clear resets LRU state
{
my $path = tmpfile();
my $map = Data::HashMap::Shared::II->new($path, 1000, 3);
shm_ii_put $map, 1, 10;
xt/stale_recovery_crash.t view on Meta::CPAN
use warnings;
use Test::More;
use POSIX qw(_exit);
use Time::HiRes qw(time);
use Data::HashMap::Shared::II;
# Regression: Pass 14 — if the process recovering a stale lock itself
# crashes mid-recovery, the lock must remain recoverable.
# Pre-fix: shm_recover_stale_lock held lock as bare WRITER_BIT (PID=0),
# which shm_pid_alive treated as always alive, causing permanent hang.
# This regression is hard to trigger deterministically — it requires a
# crash in the ~5 instruction window between CAS and seqlock-fix + release.
# Best we can do in a portable test: verify that basic operations succeed
# after a forced SIGKILL during writes, which is the common trigger path.
use File::Temp qw(tmpnam);
my $path = tmpnam() . ".$$";
my $m = Data::HashMap::Shared::II->new($path, 1024);
( run in 1.307 second using v1.01-cache-2.11-cpan-14f38c9f855 )