Data-SortedSet-Shared
view release on metacpan or search on metacpan
* A backing file -- every process calls "new($path, $max)" on the same
path. The first to arrive creates and sizes the file (serialized by an
exclusive lock); the rest map it.
* An anonymous mapping inherited across "fork" -- create with
"new(undef, $max)" before forking; the parent and its children then
share the one mapping.
* A memfd -- create with "new_memfd($name, $max)" and hand its "memfd"
descriptor to an unrelated process (over a UNIX socket with
"SCM_RIGHTS", or while the creator is alive via "/proc/$pid/fd/$n"),
which reopens it with new_from_fd($fd).
# children populate a fork-shared set; the parent reads the result
my $z = Data::SortedSet::Shared->new(undef, 1_000_000);
for my $k (1 .. 4) {
unless (fork) { # child
$z->add($k * 1_000_000 + $_, rand) for 1 .. 1000;
exit;
}
}
lib/Data/SortedSet/Shared.pm view on Meta::CPAN
=item *
B<An anonymous mapping inherited across C<fork>> -- create with
C<< new(undef, $max) >> before forking; the parent and its children then share
the one mapping.
=item *
B<A memfd> -- create with C<< new_memfd($name, $max) >> and hand its C<memfd>
descriptor to an unrelated process (over a UNIX socket with C<SCM_RIGHTS>, or
while the creator is alive via C</proc/$pid/fd/$n>), which reopens it with
C<< new_from_fd($fd) >>.
=back
# children populate a fork-shared set; the parent reads the result
my $z = Data::SortedSet::Shared->new(undef, 1_000_000);
for my $k (1 .. 4) {
unless (fork) { # child
$z->add($k * 1_000_000 + $_, rand) for 1 .. 1000;
exit;
sortedset.h view on Meta::CPAN
#else
__asm__ volatile("" ::: "memory");
#endif
}
/* Writer word encoding: WRITER_BIT|pid when write-locked, 0 when free. */
#define SS_RWLOCK_WRITER_BIT 0x80000000U
#define SS_RWLOCK_PID_MASK 0x7FFFFFFFU
#define SS_RWLOCK_WR(pid) (SS_RWLOCK_WRITER_BIT | ((uint32_t)(pid) & SS_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 ss_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 ss_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 !ss_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 ss_recover_stale_lock(SsHandle *h, uint32_t observed_wlock) {
SsHeader *hdr = h->hdr;
sortedset.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 < SS_READER_SLOTS; i++) {
uint32_t dpid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (dpid == 0 || dpid == now_pid || ss_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);
ss_occ_set(h, i);
h->my_slot_idx = i;
return;
}
}
/* Table full -- leave my_slot_idx = UINT32_MAX so this handle takes the
sortedset.h view on Meta::CPAN
* 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 ss_recover_after_timeout(SsHandle *h) {
uint32_t val = __atomic_load_n(&h->hdr->wlock, __ATOMIC_RELAXED);
if (val >= SS_RWLOCK_WRITER_BIT) {
uint32_t pid = val & SS_RWLOCK_PID_MASK;
if (!ss_pid_alive(pid))
ss_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 ss_park(SsHandle *h) {
sortedset.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). */
ss_rdepth_dec(h);
ss_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 >= SS_RWLOCK_WRITER_BIT &&
!ss_pid_alive(cur & SS_RWLOCK_PID_MASK)) {
ss_recover_stale_lock(h, cur);
spin = 0;
continue;
}
if (__builtin_expect(spin < SS_RWLOCK_SPIN_LIMIT, 1)) {
ss_rwlock_spin_pause();
continue;
}
ss_park(h);
cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
sortedset.h view on Meta::CPAN
* crash window between acquiring the lock and storing the owner. */
uint32_t mypid = SS_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 >= SS_RWLOCK_WRITER_BIT &&
!ss_pid_alive(expected & SS_RWLOCK_PID_MASK)) {
ss_recover_stale_lock(h, expected);
spin = 0;
continue;
}
if (__builtin_expect(spin < SS_RWLOCK_SPIN_LIMIT, 1)) {
ss_rwlock_spin_pause();
continue;
}
ss_park(h);
uint32_t cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
sortedset.h view on Meta::CPAN
* instead of O(SS_READER_SLOTS). */
for (uint32_t w = 0; w < SS_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 (!ss_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 */
}
xt/corrupt_shm.t view on Meta::CPAN
# Defense-in-depth for a hostile segment: node indices and index-slot state
# bytes live in peer-writable shared memory. These tests corrupt the mapping
# through the backing file (MAP_SHARED: immediately visible) and then call the
# write paths in a child process:
# * F3: a poisoned B+tree child pointer must stop the write paths (insert,
# delete, merge, underflow) instead of a wild read/write under the lock.
# * F4: an all-occupied member index must bound the open-addressing probe
# (and the backward-shift delete scan) by the table size instead of
# spinning forever under the lock.
# * F6: add_many must keep the rows array alive across element magic.
# The children run under a parent-side watchdog: the unguarded code spins in
# pure C, where Perl signal handlers never fire, so alarm() cannot be used.
# Shared-memory layout constants (must match sortedset.h; the geometry is
# cross-checked against stats() at runtime).
use constant {
HEADER_SIZE => 256,
READER_SLOTS => 1024,
SLOT_SIZE => 16,
OCC_BYTES => 128,
xt/memfd_xproc.t view on Meta::CPAN
# An unrelated process accesses a memfd-backed set it did NOT inherit, via
# /proc/<creator-pid>/fd/<n> -- the cross-process sharing memfd exists for.
pipe(my $R, my $W) or die "pipe: $!";
my $pid = fork // die "fork: $!";
if (!$pid) { # creator builds the set AFTER fork
close $R;
my $z = Data::SortedSet::Shared->new_memfd('xproc', 1000);
$z->add($_, $_ + 0.5) for 1 .. 20; # member k, score k+0.5 -> order is 1..20
syswrite $W, $$ . ' ' . $z->memfd . "\n";
close $W;
select undef, undef, undef, 5; # stay alive so /proc/$$/fd/N persists
_exit(0);
}
close $W;
my ($cpid, $cfd) = split ' ', scalar(<$R>);
open my $fh, '+<', "/proc/$cpid/fd/$cfd" or die "open /proc/$cpid/fd/$cfd: $!";
my $z2 = Data::SortedSet::Shared->new_from_fd(fileno $fh);
is $z2->count, 20, "unrelated process sees the creator's 20 members";
my @m;
$z2->each(sub { push @m, $_[0] });
( run in 3.031 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )