view release on metacpan or  search on metacpan

hiertimingwheel.h  view on Meta::CPAN

/*
 * hiertimingwheel.h -- Shared-memory hierarchical timing wheel for Linux
 *
 * O(1) timer scheduling at any delay: a hierarchical (Varghese-Lauck) timing
 * wheel of num_levels cascading wheels, each with num_slots (=S) buckets.  A
 * level-k slot spans S^k ticks, so the structure schedules any delay in
 * [1, S^num_levels).  Scheduling and cancelling a timer are O(1); a far-future
 * timer waits in a coarse level and cascades down to finer levels as its time
 * approaches, so it is touched only once per coarse tick (versus a single-level
 * wheel that revisits it every rotation).  The wheels live in a shared mapping so
 * several processes schedule into and advance one clock; a write-preferring futex
 * rwlock with reader-slot dead-process recovery guards mutation.  Each timer
 * carries an arbitrary 64-bit payload returned when it fires.
 *
 * Layout: Header -> reader_slots[1024] -> occ bitmap -> buckets[num_levels*num_slots] -> timers[capacity]
 */

#ifndef HW_H
#define HW_H

#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
#include <limits.h>
#include <signal.h>
#include <stdio.h>
#include <math.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/file.h>
#include <sys/syscall.h>
#include <linux/futex.h>
#include <pthread.h>

#if defined(__BYTE_ORDER__) && __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__
#error "hiertimingwheel.h: requires little-endian architecture"
#endif


/* ================================================================
 * Constants
 * ================================================================ */

#define HW_MAGIC        0x4C575448  /* HierTimingWheel */
#define HW_VERSION      2            /* 2: added the occupancy bitmap region (layout change) */
#define HW_ERR_BUFLEN   256
#ifndef HW_READER_SLOTS
#define HW_READER_SLOTS 1024         /* max concurrent reader processes for dead-process recovery */
#endif
/* Occupancy bitmap: one bit per reader slot, set when a process claims a slot and
 * cleared on clean release.  A writer scans these HW_OCC_WORDS words to visit
 * only OCCUPIED slots (O(words + live readers)) instead of all HW_READER_SLOTS. */
#define HW_OCC_WORDS    (((HW_READER_SLOTS) + 63) / 64)   /* 16 for 1024 slots */
#define HW_OCC_BYTES    ((uint64_t)HW_OCC_WORDS * 8)       /* 128 bytes */
#define HW_MIN_SLOTS    2
#define HW_MAX_SLOTS    0x10000U      /* 2^16 slots per level */
#define HW_MIN_LEVELS   1
#define HW_MAX_LEVELS   16            /* number of cascading wheels */
#define HW_MIN_CAP      1
#define HW_MAX_CAP      0x1000000U    /* 2^24 concurrent timers (index fits uint32, < HW_NIL) */
#define HW_NIL          0xFFFFFFFFU   /* empty list link / free-list terminator */

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

hiertimingwheel.h  view on Meta::CPAN

        HW_ERR("ftruncate: %s", strerror(errno)); close(fd); return NULL;
    }
    (void)fcntl(fd, F_ADD_SEALS, F_SEAL_SHRINK | F_SEAL_GROW);
    void *base = mmap(NULL, (size_t)total, PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);
    if (base == MAP_FAILED) { HW_ERR("mmap: %s", strerror(errno)); close(fd); return NULL; }
    hw_init_header(base, (uint32_t)num_slots, (uint32_t)num_levels, (uint32_t)capacity, total);
    return hw_setup(base, (size_t)total, NULL, fd);
}

static HwHandle *hw_open_fd(int fd, char *errbuf) {
    if (errbuf) errbuf[0] = '\0';
    struct stat st;
    if (fstat(fd, &st) < 0) { HW_ERR("fstat: %s", strerror(errno)); return NULL; }
    if ((uint64_t)st.st_size < sizeof(HwHeader)) { HW_ERR("too small"); return NULL; }
    size_t ms = (size_t)st.st_size;
    void *base = mmap(NULL, ms, PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);
    if (base == MAP_FAILED) { HW_ERR("mmap: %s", strerror(errno)); return NULL; }
    if (!hw_validate_header((HwHeader *)base, (uint64_t)st.st_size)) {
        HW_ERR("invalid hierarchical timing-wheel table"); munmap(base, ms); return NULL;
    }
    int myfd = fcntl(fd, F_DUPFD_CLOEXEC, 0);
    if (myfd < 0) { HW_ERR("fcntl: %s", strerror(errno)); munmap(base, ms); return NULL; }
    return hw_setup(base, ms, NULL, myfd);
}

static void hw_destroy(HwHandle *h) {
    if (!h) return;
    /* Release our reader slot on clean teardown (else short-lived-reader churn
     * exhausts the slot table); skip if a read lock is still held (rdepth>0). */
    if (h->reader_slots && h->my_slot_idx != UINT32_MAX && h->cached_pid &&
        h->cached_fork_gen == __atomic_load_n(&hw_fork_gen, __ATOMIC_RELAXED) &&
        __atomic_load_n(&h->reader_slots[h->my_slot_idx].rdepth, __ATOMIC_ACQUIRE) == 0) {
        /* Clear our occ bit BEFORE freeing the slot: we still own the pid so no
         * claimant can take the slot mid-clear, and rdepth==0 so no writer needs
         * to see us.  (A crash skips this -> the bit is reclaimed lazily by a
         * writer scan / re-claim, same as the pid.) */
        hw_occ_clear(h, h->my_slot_idx);
        uint32_t expected = h->cached_pid;
        __atomic_compare_exchange_n(&h->reader_slots[h->my_slot_idx].pid,
                &expected, 0, 0, __ATOMIC_RELEASE, __ATOMIC_RELAXED);
    }
    if (h->backing_fd >= 0) close(h->backing_fd);
    if (h->base) munmap(h->base, h->mmap_size);
    free(h->path);
    free(h);
}

static inline int hw_msync(HwHandle *h) {
    if (!h || !h->base) return 0;
    return msync(h->base, h->mmap_size, MS_SYNC);
}

/* ================================================================
 * Hierarchical timing-wheel operations (callers hold the lock)
 *
 * num_levels cascading wheels of num_slots (=S) buckets each; a level-k slot
 * spans tick[k]=S^k ticks, so the whole structure schedules any delay in
 * [1, S^num_levels).  A timer stores its absolute expiry; it is placed in the
 * lowest level whose range covers its remaining delay, at slot (expiry/tick[k])%S.
 * On each tick level 0 fires its current slot; when level 0 wraps, the next
 * level's now-current slot cascades down -- its timers are re-binned into finer
 * levels by remaining delay -- recursively up the levels.  Timers live in a fixed
 * pool with a free list; each bucket is a doubly-linked list (prev/next) so
 * cancellation is O(1).  Every timer index and bucket index read from shared
 * memory is bounds checked, so a corrupt link can never drive an OOB access.
 * ================================================================ */

/* unlink timer `t` from its current bucket's doubly-linked list */
static void hw_unlink(HwHandle *h, uint32_t t) {
    HwTimer *tm = hw_timer(h, t);
    uint32_t pv = tm->prev, nx = tm->next;
    if (HW_TIMER_OK(h, pv)) hw_timer(h, pv)->next = nx;
    else if (HW_BUCKET_OK(h, tm->bucket)) hw_slots(h)[tm->bucket] = nx;   /* t was the head */
    if (HW_TIMER_OK(h, nx)) hw_timer(h, nx)->prev = pv;
}

/* return timer `t` to the free list (caller has already unlinked it) */
static void hw_free(HwHandle *h, uint32_t t) {
    HwTimer *tm = hw_timer(h, t);
    tm->state  &= ~1u;   /* clear active bit; keep generation so the next reuse of this slot gets a fresh one */
    tm->prev   = HW_NIL;
    tm->bucket = HW_NIL;
    tm->next   = h->hdr->free_head;
    h->hdr->free_head = t;
}

/* link an already-populated timer `t` into flat bucket `b` (prepend) */
static void hw_link(HwHandle *h, uint32_t t, uint64_t b) {
    HwTimer *tm = hw_timer(h, t);
    tm->bucket = (uint32_t)b;
    tm->prev   = HW_NIL;
    uint32_t head = hw_slots(h)[b];
    tm->next = head;
    if (HW_TIMER_OK(h, head)) hw_timer(h, head)->prev = t;
    hw_slots(h)[b] = t;
}

/* choose the flat bucket for a timer whose absolute expiry is E, given the
 * current time now: the lowest level k with (E-now) < tick[k+1], slot
 * (E/tick[k])%S.  A remaining delay of 0 lands in level 0's current slot. */
static uint64_t hw_bucket_for(HwHandle *h, uint64_t E, uint64_t now) {
    uint64_t delay = (E > now) ? E - now : 0;
    uint32_t S = h->num_slots, L = h->num_levels, k = 0;
    while (k + 1 < L && delay >= h->tick[k + 1]) k++;      /* lowest level covering `delay` */
    uint64_t slot = (E / h->tick[k]) % S;
    return (uint64_t)k * S + slot;                         /* flat bucket index */
}

/* schedule a timer to fire in `delay` ticks (>= 1); returns its id, or -1 if the
 * pool is full or the delay exceeds the wheel's range.  (caller holds wrlock) */
static int64_t hw_add_locked(HwHandle *h, uint64_t delay, uint64_t payload) {
    if (h->num_slots == 0 || h->num_levels == 0 || h->capacity == 0) return -1;
    if (delay < 1) delay = 1;                              /* minimum effective delay is one tick */
    if (delay >= h->tick[h->num_levels]) return -2;        /* beyond S^num_levels: not schedulable */
    uint32_t t = h->hdr->free_head;
    if (!HW_TIMER_OK(h, t)) return -1;                     /* full (or corrupt free head) */
    if (hw_timer(h, t)->state & 1u) return -1;             /* Layer B: free head points at an active timer */
    h->hdr->free_head = hw_timer(h, t)->next;              /* pop the free list */

    uint64_t now = h->hdr->now, E = now + delay;
    HwTimer *tm = hw_timer(h, t);
    tm->payload = payload;
    tm->expiry  = E;
    uint32_t gen = (tm->state >> 1) + 1;                   /* bump generation for this reuse */
    tm->state   = (gen << 1) | 1u;                         /* low bit = active, upper bits = generation */
    hw_link(h, t, hw_bucket_for(h, E, now));
    h->hdr->count++;
    return ((int64_t)gen << 32) | (int64_t)t;             /* id encodes the generation so cancel() can reject a reused slot */
}

/* cancel timer `t`; returns 1 if it was active and is now removed, else 0.
 * (caller holds the write lock) */
static int hw_cancel_locked(HwHandle *h, uint64_t id) {
    uint32_t idx = (uint32_t)(id & 0xFFFFFFFFu);
    uint32_t gen = (uint32_t)((id >> 32) & 0x7FFFFFFFu);
    if (!HW_TIMER_OK(h, idx)) return 0;
    HwTimer *tm = hw_timer(h, idx);
    if (!(tm->state & 1u)) return 0;                       /* free / already fired */
    if ((tm->state >> 1) != gen) return 0;                 /* stale id: this slot was reused by a later timer */
    hw_unlink(h, idx);
    hw_free(h, idx);
    if (h->hdr->count) h->hdr->count--;
    return 1;
}

/* re-bin every timer in level-k's now-current bucket into a finer level (called
 * when `now` has just reached a multiple of tick[k]).  If level k also wrapped,
 * cascade level k+1 first (top-down) so higher levels feed into this one. */
static void hw_cascade(HwHandle *h, uint32_t k) {
    uint32_t S = h->num_slots;
    uint64_t slot = (h->hdr->now / h->tick[k]) % S;
    if (slot == 0 && k + 1 < h->num_levels) hw_cascade(h, k + 1);   /* level k wrapped too */
    uint64_t b = (uint64_t)k * S + slot;
    uint32_t t = hw_slots(h)[b];
    hw_slots(h)[b] = HW_NIL;                               /* detach the whole list */
    uint64_t guard = 0;
    while (HW_TIMER_OK(h, t) && guard++ <= (uint64_t)h->capacity) {
        HwTimer *tm = hw_timer(h, t);
        uint32_t nx = tm->next;
        if (tm->state & 1u)                                /* re-insert into a lower level (or level-0 current slot) */
            hw_link(h, t, hw_bucket_for(h, tm->expiry, h->hdr->now));
        t = nx;
    }
}

/* advance the wheel by `ticks`, collecting fired payloads into out[] (capped at
 * out_cap); returns the number fired.  (caller holds the write lock) */
static uint64_t hw_advance_locked(HwHandle *h, uint64_t ticks, uint64_t *out, uint64_t out_cap) {
    uint64_t fired = 0;
    uint32_t S = h->num_slots;
    if (S == 0 || h->num_levels == 0) return 0;
    for (uint64_t j = 0; j < ticks; j++) {
        h->hdr->now++;
        uint64_t now = h->hdr->now;
        if (now % S == 0 && h->num_levels > 1)             /* level 0 wrapped -> cascade higher levels down */
            hw_cascade(h, 1);
        uint64_t b = now % S;                              /* level-0 current slot (tick[0]==1) */
        uint32_t t = hw_slots(h)[b];
        uint64_t guard = 0;
        while (HW_TIMER_OK(h, t) && guard++ <= (uint64_t)h->capacity) {
            HwTimer *tm = hw_timer(h, t);
            if (!(tm->state & 1u)) break;                  /* Layer B: corrupt link into a freed node */
            uint32_t nx = tm->next;
            hw_unlink(h, t);                               /* every timer in level-0's current slot is due now */
            hw_free(h, t);
            if (h->hdr->count) h->hdr->count--;
            if (fired < out_cap) out[fired++] = tm->payload;   /* cap BOTH the write and the count */
            t = nx;
        }
    }
    return fired;
}

/* reset to an empty wheel: rethread the free list, clear the buckets, reset time.
 * (caller holds the write lock) */
static inline void hw_clear_locked(HwHandle *h) {
    uint64_t nb = hw_num_bucket(h), cap = h->capacity;
    uint64_t smax = (h->slots_off < h->mmap_size) ? (h->mmap_size - h->slots_off) / sizeof(uint32_t) : 0;
    if (nb > smax) nb = smax;                              /* Layer B */
    uint64_t tmax = hw_timers_max(h);
    if (cap > tmax) cap = tmax;
    uint32_t *slots = hw_slots(h);
    for (uint64_t s = 0; s < nb; s++) slots[s] = HW_NIL;
    for (uint64_t i = 0; i < cap; i++) {
        HwTimer *tm = hw_timer(h, i);
        tm->next   = (i + 1 < cap) ? (uint32_t)(i + 1) : HW_NIL;
        tm->prev   = HW_NIL;
        tm->bucket = HW_NIL;
        tm->state  &= ~1u;   /* clear active bit; keep generation so ids from before clear() stay distinguishable (no clear-then-reuse ABA) */
    }
    h->hdr->now = 0;
    h->hdr->count = 0;
    h->hdr->free_head = cap ? 0 : HW_NIL;
}

#endif /* HW_H */