Data-Buffer-Shared
view release on metacpan or search on metacpan
"$buf->get_raw(4, 4)" returns bytes 4..7, which is the upper half of
element 0 and the lower half of element 1, not elements 4..7. Multiply by
the element size to address elements. Both are bounds-checked against the
data area and croak rather than run past it.
Zero-copy:
my $sv = $buf->as_scalar; # mmap-aliased read-only scalar ref
The returned scalar aliases the mapped bytes directly (no copy) and holds
a reference to the buffer so the mapping stays alive while it is in use.
Cross-process notification (all variants):
my $efd = $buf->create_eventfd; # create + attach an eventfd, returns the fd
$buf->attach_eventfd($fd); # attach an already-open eventfd
my $efd = $buf->eventfd; # current eventfd, or undef if none
$buf->notify; # signal (eventfd write)
my $n = $buf->wait_notify; # drain the counter, non-blocking (undef if 0)
These are a thin wrapper over an eventfd(2) descriptor stored in the
#include "buf_u32.h"
#include "buf_i64.h"
#include "buf_u64.h"
#include "buf_f32.h"
#include "buf_f64.h"
#include "buf_str.h"
#include "XSParseKeyword.h"
/* ---- as_scalar magic: prevent use-after-free by preventing buffer DESTROY
* while the returned scalar ref is alive. We attach magic to the inner SV that
* holds a reference to the buffer's underlying handle -- the REFERENT (SvRV),
* not the caller's RV container: `undef $buf` clears the container in place and
* releases the handle, which would free the mapping while the inner SV still
* aliases it. When the inner SV is freed, the magic destructor releases the
* reference (deferring DESTROY until both the caller and the scalar are gone). ---- */
static int buf_scalar_magic_free(pTHX_ SV *sv, MAGIC *mg) {
PERL_UNUSED_ARG(sv);
if (mg->mg_obj) SvREFCNT_dec(mg->mg_obj);
return 0;
buf_generic.h view on Meta::CPAN
* is DISTRIBUTED across per-process reader slots: each slot's `rdepth` is that
* process's entire contribution. A reader publishes its presence in its own slot
* then re-checks the writer word; a writer publishes the writer word then scans
* every slot until all live readers' rdepth reach 0. Sequentially-consistent
* store+load on each side (a Dekker handshake) gives mutual exclusion. A crashed
* reader is recovered by clearing its one slot (CAS pid->0) -- no second counter
* to strand, no orphaned +1, no quiescent force-reset -- so sustained read
* traffic can never starve a writer. Write-preference is inherent in the gate
* (new readers see wlock!=0 and yield). */
/* 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 buf_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';
}
static inline int buf_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 !buf_pid_is_zombie(pid); /* kill() also succeeds for a zombie -> treat as dead */
}
/* ---- Per-process slot lifecycle (dead-reader recovery) ----
* Each process claims one BufReaderSlot lazily on first lock op so that
* its contribution to the shared rwlock counter can be reclaimed by other
* processes if it dies (SIGKILL'd worker no longer pins the counter). */
static uint32_t buf_fork_gen = 0;
static pthread_once_t buf_atfork_once = PTHREAD_ONCE_INIT;
buf_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 < BUF_READER_SLOTS; i++) {
uint32_t dpid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (dpid == 0 || dpid == now_pid || buf_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);
buf_occ_set(h, i);
h->my_slot_idx = i;
return;
}
}
/* Table full -- leave my_slot_idx = UINT32_MAX so this handle takes the
buf_generic.h view on Meta::CPAN
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 buf_rdepth_dec(BufHandle *h) {
/* Drop ONLY what this handle actually published. An unbalanced unlock_rd
* (more unlocks than locks) would otherwise decrement a slot rdepth of 0,
* wrapping it to UINT32_MAX: the owning pid is alive, so dead-reader
* recovery never fires and every writer on the buffer -- in every process --
* blocks forever inside the drain futex, uninterruptible by Perl signals. */
if (h->rd_held == 0) return;
h->rd_held--;
/* Peel slotless holds first. A reader that published to slotless_rdepth and
* then claimed a freed slot mid-hold would otherwise decrement the SLOT,
* underflowing its rdepth and stranding slotless_rdepth at a nonzero value --
* which nothing ever recovers, so every future writer would block forever. */
if (h->slotless_held > 0) {
h->slotless_held--;
buf_generic.h view on Meta::CPAN
static const struct timespec buf_lock_timeout = { BUF_LOCK_TIMEOUT_SEC, 0 };
/* 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 buf_recover_after_timeout(BufHandle *h) {
BufHeader *hdr = h->hdr;
uint32_t val = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (val >= BUF_RWLOCK_WRITER_BIT) {
uint32_t pid = val & BUF_RWLOCK_PID_MASK;
if (!buf_pid_alive(pid))
buf_recover_stale_lock(hdr, val);
}
}
static inline void buf_rwlock_rdlock(BufHandle *h) {
BufHeader *hdr = h->hdr;
buf_claim_reader_slot(h);
for (int spin = 0; ; spin++) {
uint32_t cur = __atomic_load_n(&hdr->wlock, __ATOMIC_ACQUIRE);
if (cur == 0) {
buf_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). */
buf_rdepth_dec(h);
buf_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 >= BUF_RWLOCK_WRITER_BIT &&
!buf_pid_alive(cur & BUF_RWLOCK_PID_MASK)) {
buf_recover_stale_lock(hdr, cur);
spin = 0;
continue;
}
if (__builtin_expect(spin < BUF_RWLOCK_SPIN_LIMIT, 1)) {
buf_spin_pause();
continue;
}
buf_park(h);
cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
buf_generic.h view on Meta::CPAN
buf_claim_reader_slot(h);
uint32_t mypid = BUF_RWLOCK_WR((uint32_t)getpid());
/* 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 >= BUF_RWLOCK_WRITER_BIT &&
!buf_pid_alive(expected & BUF_RWLOCK_PID_MASK)) {
buf_recover_stale_lock(hdr, expected);
spin = 0;
continue;
}
if (__builtin_expect(spin < BUF_RWLOCK_SPIN_LIMIT, 1)) {
buf_spin_pause();
continue;
}
buf_park(h);
uint32_t cur = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
buf_generic.h view on Meta::CPAN
* instead of O(BUF_READER_SLOTS). */
for (uint32_t w = 0; w < BUF_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 (!buf_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;
if (__atomic_compare_exchange_n(&h->reader_slots[i].pid, &ep, 0,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED))
__atomic_add_fetch(&hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
continue;
}
busy = 1; /* live reader still holding */
buf_generic.h view on Meta::CPAN
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)) {
buf_spin_pause();
spin++;
continue;
}
uint32_t val = __atomic_load_n(&hdr->wlock, __ATOMIC_RELAXED);
if (val >= BUF_RWLOCK_WRITER_BIT) {
uint32_t pid = val & BUF_RWLOCK_PID_MASK;
if (!buf_pid_alive(pid)) {
buf_recover_stale_lock(hdr, val);
spin = 0;
continue;
}
}
struct timespec ts = {0, 1000000};
nanosleep(&ts, NULL);
spin = 0;
}
}
buf_generic.h view on Meta::CPAN
uint64_t val = 0;
if (read(h->efd, &val, sizeof(val)) != sizeof(val)) return -1;
return (int64_t)val;
}
static void buf_close_map(BufHandle *h) {
if (!h) return;
/* Release any lock this handle still holds BEFORE freeing it. Destroying a
* handle mid-lock (e.g. `$b->lock_rd; undef $b;`) otherwise pins the reader
* slot with a LIVE pid and rdepth > 0, or leaves wlock set to a live pid --
* and because that pid is alive, dead-owner recovery never fires, so every
* other process starves until this one exits. rdepth is per-process and
* shared by all handles, so we can only drop what THIS handle owes. */
if (h->hdr) {
/* Release only locks THIS process took. A forked child inherits
* rd_held/wr_locked verbatim, but those holds were published by the
* parent (its slot rdepth, its pid in wlock): dropping them here
* would unlock the parent's live critical section. Same owner test
* as the slot-release guard below; cached_pid/cached_fork_gen are
* recorded by buf_claim_reader_slot, which every lock op runs, so a
* nonzero rd_held/wr_locked implies they are set. */
lib/Data/Buffer/Shared.pm view on Meta::CPAN
C<< $buf->get_raw(4, 4) >> returns bytes 4..7, which is the upper half of
element 0 and the lower half of element 1, not elements 4..7. Multiply by the
element size to address elements. Both are bounds-checked against the data area
and croak rather than run past it.
Zero-copy:
my $sv = $buf->as_scalar; # mmap-aliased read-only scalar ref
The returned scalar aliases the mapped bytes directly (no copy) and holds a
reference to the buffer so the mapping stays alive while it is in use.
Cross-process notification (all variants):
my $efd = $buf->create_eventfd; # create + attach an eventfd, returns the fd
$buf->attach_eventfd($fd); # attach an already-open eventfd
my $efd = $buf->eventfd; # current eventfd, or undef if none
$buf->notify; # signal (eventfd write)
my $n = $buf->wait_notify; # drain the counter, non-blocking (undef if 0)
These are a thin wrapper over an C<eventfd(2)> descriptor stored in the handle,
t/09-review-fixes.t view on Meta::CPAN
my $buf = Data::Buffer::Shared::Str->new_anon(5, 16);
$buf->set(0, "hello");
$buf->lock_wr;
is($buf->get(0), "hello", 'str get under lock_wr');
my @vals = $buf->slice(0, 1);
is($vals[0], "hello", 'str slice under lock_wr');
$buf->unlock_wr;
}
# === as_scalar keeps buffer alive (prevents use-after-free) ===
{
my $ref;
{
my $buf = Data::Buffer::Shared::I64->new_anon(10);
$buf->set(0, 12345);
$ref = $buf->as_scalar;
# $buf goes out of scope here — but magic ref prevents DESTROY
}
# buffer should still be alive because $ref holds a backref
my @vals = unpack("q<", $$ref);
is($vals[0], 12345, 'as_scalar keeps buffer alive after scope exit');
}
done_testing;
t/14-unbalanced-unlock.t view on Meta::CPAN
use strict;
use warnings;
use Test::More;
use File::Temp qw(tempdir);
use Data::Buffer::Shared::I64;
# An unbalanced unlock_rd (more unlocks than locks) must not decrement a reader
# slot this handle never incremented. It used to: the slot's rdepth wrapped from
# 0 to UINT32_MAX, and because the owning pid is alive, dead-reader recovery never
# fires -- so every writer on the buffer, in every process, blocked forever inside
# the drain futex. The hang is in a syscall under XS, so even Perl's alarm cannot
# break it; the process has to be killed.
my $dir = tempdir( CLEANUP => 1 );
my $p = "$dir/unbalanced.i64";
my $b = Data::Buffer::Shared::I64->new( $p, 128 );
# Claim a reader slot and release it cleanly.
xt/lock_leak_on_destroy.t view on Meta::CPAN
#!/usr/bin/perl
# Regression: destroying a handle that still holds a lock must not strand it.
#
# buf_close_map released the process's reader slot only when rdepth == 0, and
# never released a held write lock. So `$b->lock_rd; undef $b;` left the slot
# pinned with a LIVE pid and rdepth > 0 -- and because that pid is alive,
# dead-owner recovery never fires, so every other process's lock_wr starved
# until this process exited.
#
# rdepth is per-process and shared by all handles, so the handle now tracks how
# much of it IT owns (rd_held) and releases exactly that on close.
#
# The waiting writer blocks inside XS (futex), which Perl's alarm cannot
# interrupt, so it runs in a child with a hard timeout from the parent.
use strict;
use warnings;
( run in 4.242 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )