Data-Pool-Shared
view release on metacpan or search on metacpan
#define POOL_VAR_STR 4
#define POOL_ALIGN8(x) (((x) + 7) & ~(uint64_t)7)
/* ================================================================
* Header (128 bytes = 2 cache lines)
* ================================================================ */
typedef struct {
/* ---- Cache line 0 (0-63): immutable after create ---- */
uint32_t magic; /* 0 */
uint32_t version; /* 4 */
uint32_t elem_size; /* 8 */
uint32_t variant_id; /* 12 */
uint64_t capacity; /* 16: number of slots */
uint64_t total_size; /* 24: total mmap size */
uint64_t data_off; /* 32: offset to slot data */
uint64_t bitmap_off; /* 40: offset to allocation bitmap */
uint64_t owners_off; /* 48: offset to per-slot owner PIDs */
uint8_t _pad0[8]; /* 56-63 */
/* ---- Cache line 1 (64-127): mutable state ---- */
uint32_t used; /* 64: allocated count (futex word) */
uint32_t waiters; /* 68: blocked on alloc */
uint8_t _pad1[8]; /* 72-79 */
uint64_t stat_allocs; /* 80 */
uint64_t stat_frees; /* 88 */
uint64_t stat_waits; /* 96 */
uint64_t stat_timeouts; /* 104 */
uint64_t stat_recoveries;/* 112 */
uint8_t _pad2[8]; /* 120-127 */
} PoolHeader;
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
_Static_assert(sizeof(PoolHeader) == 128, "PoolHeader must be 128 bytes");
#endif
/* ================================================================
* Process-local handle
* ================================================================ */
typedef struct {
PoolHeader *hdr;
uint64_t *bitmap;
uint32_t *owners;
uint8_t *data;
size_t mmap_size;
uint32_t bitmap_words;
uint64_t capacity; /* cached-at-attach geometry (peer may corrupt hdr->capacity) */
uint32_t elem_size; /* cached-at-attach geometry (peer may corrupt hdr->elem_size) */
char *path;
int notify_fd;
int backing_fd;
uint32_t scan_hint;
} PoolHandle;
/* ================================================================
* Utility
* ================================================================ */
/* 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 pool_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 pool_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 !pool_pid_is_zombie(pid); /* kill() also succeeds for a zombie -> treat as dead */
}
static inline void pool_make_deadline(double timeout, struct timespec *deadline) {
clock_gettime(CLOCK_MONOTONIC, deadline);
if (!(timeout < 1e9)) timeout = 1e9; /* clamp Inf/NaN/huge: avoid UB (time_t) cast -> instant spurious timeout */
deadline->tv_sec += (time_t)timeout;
deadline->tv_nsec += (long)((timeout - (double)(time_t)timeout) * 1e9);
if (deadline->tv_nsec >= 1000000000L) {
deadline->tv_sec++;
deadline->tv_nsec -= 1000000000L;
}
}
static inline int pool_remaining_time(const struct timespec *deadline,
struct timespec *remaining) {
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
remaining->tv_sec = deadline->tv_sec - now.tv_sec;
remaining->tv_nsec = deadline->tv_nsec - now.tv_nsec;
if (remaining->tv_nsec < 0) {
remaining->tv_sec--;
remaining->tv_nsec += 1000000000L;
}
return remaining->tv_sec >= 0;
}
/* ================================================================
* Slot access
* ================================================================ */
static inline uint8_t *pool_slot_ptr(PoolHandle *h, uint64_t slot) {
return h->data + slot * h->elem_size;
}
static inline int pool_is_allocated(PoolHandle *h, uint64_t slot) {
uint32_t widx = (uint32_t)(slot / 64);
int bit = (int)(slot % 64);
uint64_t word = __atomic_load_n(&h->bitmap[widx], __ATOMIC_RELAXED);
return (word >> bit) & 1;
}
/* ================================================================
* Allocation (lock-free bitmap scan + CAS)
* ================================================================ */
static inline int64_t pool_try_alloc(PoolHandle *h) {
uint32_t nwords = h->bitmap_words;
uint64_t cap = h->capacity;
uint32_t start = h->scan_hint;
uint32_t mypid = (uint32_t)getpid();
for (uint32_t i = 0; i < nwords; i++) {
uint32_t widx = (start + i) % nwords;
uint64_t word = __atomic_load_n(&h->bitmap[widx], __ATOMIC_RELAXED);
while (word != ~(uint64_t)0) {
int bit = __builtin_ctzll(~word);
uint64_t slot = (uint64_t)widx * 64 + bit;
if (timeout == 0) {
for (uint32_t i = 0; i < count; i++) {
int64_t slot = pool_try_alloc(h);
if (slot < 0) {
if (i > 0) pool_free_n(h, out, i);
return 0;
}
out[i] = (uint64_t)slot;
}
return 1;
}
/* count > capacity can never be satisfied: fail fast rather than grab the
* whole pool and block forever. */
if ((uint64_t)count > h->hdr->capacity) return 0;
PoolHeader *hdr = h->hdr;
struct timespec deadline, remaining;
int has_deadline = (timeout > 0);
if (has_deadline) pool_make_deadline(timeout, &deadline);
/* All-or-nothing with retry, releasing any partial claim BEFORE waiting:
* holding it across the wait let two over-demanding alloc_n callers deadlock. */
uint32_t backoff = 0;
for (;;) {
uint32_t got = 0;
for (; got < count; got++) {
int64_t s = pool_try_alloc(h);
if (s < 0) break;
out[got] = (uint64_t)s;
}
if (got == count) return 1;
if (got > 0) pool_free_n(h, out, got); /* don't hold across the wait */
if (has_deadline && !pool_remaining_time(&deadline, &remaining)) {
__atomic_add_fetch(&hdr->stat_timeouts, 1, __ATOMIC_RELAXED);
return 0;
}
/* Jittered (pid-seeded) backoff desyncs competing alloc_n callers,
* breaking the symmetric grab/release livelock. */
{
long us = (long)(((getpid() & 0x7u) + backoff) % 40u) * 25 + 25; /* 25..1000 us */
if (backoff < 40) backoff++;
struct timespec bt = { 0, us * 1000L };
nanosleep(&bt, NULL);
}
}
}
/* ================================================================
* Stale recovery -- CAS owner to narrow race window
* ================================================================ */
static inline uint32_t pool_recover_stale(PoolHandle *h) {
uint32_t recovered = 0;
uint64_t cap = h->capacity;
for (uint64_t slot = 0; slot < cap; slot++) {
if (!pool_is_allocated(h, slot)) continue;
uint32_t owner = __atomic_load_n(&h->owners[slot], __ATOMIC_ACQUIRE);
if (owner == 0 || pool_pid_alive(owner)) continue;
/* CAS owner from dead PID to 0 -- if it fails, slot was
* re-allocated or already recovered by another process */
if (!__atomic_compare_exchange_n(&h->owners[slot], &owner, 0,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED))
continue;
/* We now own the right to free this slot's bitmap bit.
*
* Race window: between our owner-CAS and the bitmap-CAS below,
* the bitmap word can transition via concurrent allocators (or,
* with API misuse, a free+alloc cycle on the same bit). The bit
* may be reset by an allocator before our CAS reaches it, or our
* CAS may clear a bit a fresh allocator has just claimed.
*
* Mitigation: pre-CAS owner check (narrows window) + post-CAS
* recovery accounting (always decrement used, since our CAS
* succeeded against an "expected = bit set" state -- that bit is
* gone from popcount). If post-CAS observes an owner already
* stored, an allocator's CAS landed inside our window; we restore
* the bit so their slot stays claimed. Their own used++ pairs
* with our used-- to keep the counter in sync with popcount. */
uint32_t widx = (uint32_t)(slot / 64);
int bit = (int)(slot % 64);
uint64_t mask = (uint64_t)1 << bit;
for (;;) {
uint64_t word = __atomic_load_n(&h->bitmap[widx], __ATOMIC_RELAXED);
if (!(word & mask)) break;
/* Pre-CAS: if a new allocator already populated owner, abort */
if (__atomic_load_n(&h->owners[slot], __ATOMIC_ACQUIRE) != 0)
break;
uint64_t new_word = word & ~mask;
if (__atomic_compare_exchange_n(&h->bitmap[widx], &word, new_word,
1, __ATOMIC_RELEASE, __ATOMIC_RELAXED)) {
/* Post-CAS: if owner is now non-zero, a live allocator's
* CAS landed inside our window. Restore the bit so their
* claim stays valid; their used++ already happened. */
if (__atomic_load_n(&h->owners[slot], __ATOMIC_ACQUIRE) != 0)
__atomic_or_fetch(&h->bitmap[widx], mask, __ATOMIC_RELEASE);
/* Account for the stale slot's bit we just cleared. Use a
* saturating decrement: if a prior allocator was killed
* between owner-store and used++, used may already reflect
* fewer "real" allocations than popcount suggests, and a
* plain sub would underflow on subsequent recoveries. */
uint32_t cur = __atomic_load_n(&h->hdr->used, __ATOMIC_RELAXED);
while (cur > 0 && !__atomic_compare_exchange_n(&h->hdr->used,
&cur, cur - 1, 1, __ATOMIC_RELEASE, __ATOMIC_RELAXED))
; /* CAS failed: cur reloaded with current value; retry */
__atomic_add_fetch(&h->hdr->stat_frees, 1, __ATOMIC_RELAXED);
/* StoreLoad barrier: see pool_free_slot. */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&h->hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &h->hdr->used, FUTEX_WAKE, 1, NULL, NULL, 0);
recovered++;
break;
}
}
( run in 1.156 second using v1.01-cache-2.11-cpan-14f38c9f855 )