view release on metacpan or  search on metacpan

pool.h  view on Meta::CPAN

    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 (slot >= cap) break;

            uint64_t new_word = word | ((uint64_t)1 << bit);
            if (__atomic_compare_exchange_n(&h->bitmap[widx], &word, new_word,
                    1, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
                __atomic_store_n(&h->owners[slot], mypid, __ATOMIC_RELAXED);
                memset(pool_slot_ptr(h, slot), 0, h->elem_size);