view release on metacpan or  search on metacpan

heap.h  view on Meta::CPAN

                if (!heap_pid_alive(pid)) {
                    if (__atomic_compare_exchange_n(&hdr->mutex, &cur, 0,
                            0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED)) {
                        __atomic_add_fetch(&hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
                        /* Wake one waiter so recovery latency is not bounded by the 2s timeout. */
                        if (__atomic_load_n(&hdr->mutex_waiters, __ATOMIC_RELAXED) > 0)
                            syscall(SYS_futex, &hdr->mutex, FUTEX_WAKE, 1, NULL, NULL, 0);
                    }
                }
            }
        }
        __atomic_sub_fetch(&hdr->mutex_waiters, 1, __ATOMIC_RELAXED);
        spin = 0;
    }
}

static inline void heap_mutex_unlock(HeapHeader *hdr) {
    __atomic_store_n(&hdr->mutex, 0, __ATOMIC_RELEASE);
    if (__atomic_load_n(&hdr->mutex_waiters, __ATOMIC_RELAXED) > 0)
        syscall(SYS_futex, &hdr->mutex, FUTEX_WAKE, 1, NULL, NULL, 0);
}

/* ================================================================
 * Heap operations (must hold mutex)
 * ================================================================ */

static inline void heap_swap(HeapEntry *a, HeapEntry *b) {
    HeapEntry t = *a; *a = *b; *b = t;
}

static inline void heap_sift_up(HeapEntry *data, uint32_t idx) {
    while (idx > 0) {
        uint32_t parent = (idx - 1) / 2;
        if (data[parent].priority <= data[idx].priority) break;
        heap_swap(&data[parent], &data[idx]);
        idx = parent;
    }
}

static inline void heap_sift_down(HeapEntry *data, uint32_t size, uint32_t idx) {
    while (1) {
        uint32_t smallest = idx;
        uint64_t left = (uint64_t)idx * 2 + 1;   /* uint64: 2*idx overflows uint32 near 2^31 */
        uint64_t right = (uint64_t)idx * 2 + 2;
        if (left < size && data[left].priority < data[smallest].priority)
            smallest = (uint32_t)left;
        if (right < size && data[right].priority < data[smallest].priority)
            smallest = (uint32_t)right;
        if (smallest == idx) break;
        heap_swap(&data[idx], &data[smallest]);
        idx = smallest;
    }
}

/* ================================================================
 * Public API
 * ================================================================ */

static inline void heap_make_deadline(double t, struct timespec *dl) {
    clock_gettime(CLOCK_MONOTONIC, dl);
    if (!(t < 1e9)) t = 1e9; /* clamp Inf/NaN/huge: avoid UB (time_t) cast -> instant spurious timeout */
    dl->tv_sec += (time_t)t;
    dl->tv_nsec += (long)((t - (double)(time_t)t) * 1e9);
    if (dl->tv_nsec >= 1000000000L) { dl->tv_sec++; dl->tv_nsec -= 1000000000L; }
}

static inline int heap_remaining(const struct timespec *dl, struct timespec *rem) {
    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    rem->tv_sec = dl->tv_sec - now.tv_sec;
    rem->tv_nsec = dl->tv_nsec - now.tv_nsec;
    if (rem->tv_nsec < 0) { rem->tv_sec--; rem->tv_nsec += 1000000000L; }
    return rem->tv_sec >= 0;
}

static inline int heap_push(HeapHandle *h, int64_t priority, int64_t value) {
    HeapHeader *hdr = h->hdr;
    heap_mutex_lock(hdr);
    uint32_t cur = hdr->size;
    if (cur >= h->capacity) {
        heap_mutex_unlock(hdr);
        return 0;
    }
    uint32_t idx = cur;
    hdr->size = cur + 1;
    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.  Full barrier so our size store (under the lock) is
     * globally ordered before this waiters_pop load; pairs with the matching
     * fence in heap_pop_wait so a concurrent waiter either observes our push
     * (via the futex value re-check) or we observe its waiters_pop increment
     * -- Dekker, no lost wakeup on weakly-ordered CPUs. */
    __atomic_thread_fence(__ATOMIC_SEQ_CST);
    if (__atomic_load_n(&hdr->waiters_pop, __ATOMIC_SEQ_CST) > 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);
    uint32_t cur = hdr->size;
    if (cur == 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.  Read it once into a local and use the local throughout. */
    if (cur > h->capacity) {
        heap_mutex_unlock(hdr);
        return 0;
    }
    *out_priority = h->data[0].priority;
    *out_value = h->data[0].value;