view release on metacpan or  search on metacpan

heap.h  view on Meta::CPAN

    HeapHeader *hdr = h->hdr;
    heap_mutex_lock(hdr);
    if (hdr->size >= h->capacity) {
        heap_mutex_unlock(hdr);
        return 0;
    }
    uint32_t idx = hdr->size++;
    h->data[idx].priority = priority;
    h->data[idx].value = value;
    heap_sift_up(h->data, idx);
    __atomic_add_fetch(&hdr->stat_pushes, 1, __ATOMIC_RELAXED);
    heap_mutex_unlock(hdr);
    /* wake pop-waiters */
    if (__atomic_load_n(&hdr->waiters_pop, __ATOMIC_RELAXED) > 0)
        syscall(SYS_futex, &hdr->size, FUTEX_WAKE, 1, NULL, NULL, 0);
    return 1;
}

static inline int heap_pop(HeapHandle *h, int64_t *out_priority, int64_t *out_value) {
    HeapHeader *hdr = h->hdr;
    heap_mutex_lock(hdr);
    if (hdr->size == 0) {
        heap_mutex_unlock(hdr);
        return 0;
    }
    /* Layer B: size is read from the shared segment and used just below as an
     * array index (h->data[size]) and as the sift-down bound.  A local peer
     * with write access to the backing file can corrupt it past capacity,
     * which would drive an out-of-bounds read/write.  A valid heap always
     * keeps size <= capacity (push enforces it), so this never fires for
     * good data. */
    if (hdr->size > h->capacity) {
        heap_mutex_unlock(hdr);
        return 0;
    }
    *out_priority = h->data[0].priority;
    *out_value = h->data[0].value;
    hdr->size--;
    if (hdr->size > 0) {
        h->data[0] = h->data[hdr->size];
        heap_sift_down(h->data, hdr->size, 0);
    }
    __atomic_add_fetch(&hdr->stat_pops, 1, __ATOMIC_RELAXED);
    heap_mutex_unlock(hdr);
    return 1;
}

static inline int heap_pop_wait(HeapHandle *h, int64_t *out_p, int64_t *out_v, double timeout) {
    if (heap_pop(h, out_p, out_v)) return 1;
    if (timeout == 0) return 0;

    HeapHeader *hdr = h->hdr;
    struct timespec dl, rem;
    int has_dl = (timeout > 0);
    if (has_dl) heap_make_deadline(timeout, &dl);
    __atomic_add_fetch(&hdr->stat_waits, 1, __ATOMIC_RELAXED);

    for (;;) {
        __atomic_add_fetch(&hdr->waiters_pop, 1, __ATOMIC_RELEASE);
        uint32_t cur = __atomic_load_n(&hdr->size, __ATOMIC_ACQUIRE);
        if (cur > h->capacity) {   /* corrupt size: heap_pop can never succeed, so do not busy-spin */
            __atomic_sub_fetch(&hdr->waiters_pop, 1, __ATOMIC_RELAXED);
            return 0;
        }
        if (cur == 0) {
            struct timespec *pts = NULL;
            if (has_dl) {
                if (!heap_remaining(&dl, &rem)) {
                    __atomic_sub_fetch(&hdr->waiters_pop, 1, __ATOMIC_RELAXED);
                    __atomic_add_fetch(&hdr->stat_timeouts, 1, __ATOMIC_RELAXED);
                    return 0;
                }
                pts = &rem;
            }
            syscall(SYS_futex, &hdr->size, FUTEX_WAIT, 0, pts, NULL, 0);
        }
        __atomic_sub_fetch(&hdr->waiters_pop, 1, __ATOMIC_RELAXED);
        if (heap_pop(h, out_p, out_v)) return 1;
        if (has_dl && !heap_remaining(&dl, &rem)) {
            __atomic_add_fetch(&hdr->stat_timeouts, 1, __ATOMIC_RELAXED);
            return 0;
        }
    }
}

static inline int heap_peek(HeapHandle *h, int64_t *out_p, int64_t *out_v) {
    HeapHeader *hdr = h->hdr;
    heap_mutex_lock(hdr);
    if (hdr->size == 0) { heap_mutex_unlock(hdr); return 0; }
    *out_p = h->data[0].priority;
    *out_v = h->data[0].value;
    heap_mutex_unlock(hdr);
    return 1;
}

static inline uint32_t heap_size(HeapHandle *h) {
    return __atomic_load_n(&h->hdr->size, __ATOMIC_RELAXED);
}

/* ================================================================
 * Create / Open / Close
 * ================================================================ */

#define HEAP_ERR(fmt, ...) do { if (errbuf) snprintf(errbuf, HEAP_ERR_BUFLEN, fmt, ##__VA_ARGS__); } while(0)

static inline void heap_init_header(void *base, uint64_t total, uint64_t capacity) {
    HeapHeader *hdr = (HeapHeader *)base;
    memset(base, 0, (size_t)total);
    hdr->magic     = HEAP_MAGIC;
    hdr->version   = HEAP_VERSION;
    hdr->capacity  = capacity;
    hdr->total_size = total;
    hdr->data_off  = sizeof(HeapHeader);
    __atomic_thread_fence(__ATOMIC_SEQ_CST);
}

/* Validate a mapped header (shared by heap_create reopen and heap_open_fd). */
static inline int heap_validate_header(const HeapHeader *hdr, uint64_t file_size) {
    if (hdr->magic != HEAP_MAGIC) return 0;
    if (hdr->version != HEAP_VERSION) return 0;
    if (hdr->capacity == 0 || hdr->capacity > HEAP_MAX_CAPACITY) return 0;