Data-IntervalTree-Shared
view release on metacpan or search on metacpan
intervaltree.h view on Meta::CPAN
/*
* intervaltree.h -- Shared-memory interval tree for Linux
*
* A set of integer intervals [lo, hi] with fast overlap queries: which stored
* intervals contain a point (stabbing), or overlap a query range. Backed by an
* augmented balanced BST -- keyed by the low endpoint, each node carrying the
* maximum high endpoint of its subtree so a query prunes subtrees that end
* before it. Intervals are appended in O(1) and the balanced tree is bulk-built
* on the first query after any insert (so query recursion stays O(log n) deep
* regardless of insertion order). The intervals and tree live in a shared
* mapping so several processes build and query one index; a write-preferring
* futex rwlock with reader-slot dead-process recovery guards mutation, and
* queries take only the read lock once the tree is built.
*
* Layout: Header -> reader_slots[1024] -> nodes[capacity] -> build_idx[capacity]
*/
#ifndef IT_H
#define IT_H
#include <stddef.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 "intervaltree.h: requires little-endian architecture"
#endif
/* ================================================================
* Constants
* ================================================================ */
#define IT_MAGIC 0x52545649 /* IntervalTree */
#define IT_VERSION 2 /* 2: added the occupancy bitmap region (layout change) */
#define IT_ERR_BUFLEN (PATH_MAX + 256)
#ifndef IT_READER_SLOTS
#define IT_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 IT_OCC_WORDS words to visit
* only OCCUPIED slots (O(words + live readers)) instead of all IT_READER_SLOTS. */
#define IT_OCC_WORDS (((IT_READER_SLOTS) + 63) / 64) /* 16 for 1024 slots */
#define IT_OCC_BYTES ((uint64_t)IT_OCC_WORDS * 8) /* 128 bytes */
#define IT_MIN_CAP 1
#define IT_MAX_CAP 0x1000000U /* 2^24 intervals cap (index fits uint32, < IT_NIL) */
#define IT_NIL 0xFFFFFFFFU /* empty child / root sentinel */
/* A balanced bulk-built tree over <= 2^24 intervals is <= ~25 deep; this cap is
* a runaway guard so a Layer-B-corrupted link chain (a cycle, or a degenerate
* chain of valid indices) cannot recurse a query into a stack overflow. */
#define IT_MAX_DEPTH 96
#define IT_ERR(fmt, ...) do { if (errbuf) snprintf(errbuf, IT_ERR_BUFLEN, fmt, ##__VA_ARGS__); } while (0)
/* ================================================================
* Structs
* ================================================================ */
/* Per-process slot for dead-process recovery. In the reader-slots-only rwlock a
* reader's ENTIRE contribution to the shared lock is `rdepth` in its OWN slot --
* there is no separate shared reader counter to fall out of sync with it -- so a
* dead reader's contribution is exactly this one word, which a draining writer
* neutralises by clearing the slot's pid (the scan then ignores the slot). No
* orphaned counter can exist, so there is no quiescent force-reset and sustained
* readers cannot starve a writer. _rsv1/_rsv2 are kept only to preserve the
* 16-byte slot size across the already-released builds. */
typedef struct {
uint32_t pid; /* 0 = unclaimed */
uint32_t rdepth; /* read-locks THIS process currently holds (recursion-safe) */
uint32_t _rsv1; /* reserved (was waiters_parked); unused, kept for layout size */
uint32_t _rsv2; /* reserved (was writers_parked); unused, kept for layout size */
} ItReaderSlot;
struct ItHeader {
uint32_t magic, version; /* 0,4 */
uint32_t capacity; /* 8 max intervals */
uint32_t _pad0; /* 12 */
uint64_t count; /* 16 intervals inserted */
uint32_t root; /* 24 root node index, or IT_NIL when empty */
uint32_t dirty; /* 28 1 if the tree needs a (re)build before querying */
uint64_t nodes_off; /* 32 offset of the node array */
uint64_t idx_off; /* 40 offset of the build scratch (capacity uint32) */
uint64_t total_size; /* 48 */
uint64_t reader_slots_off; /* 56 */
uint32_t wlock; /* 64 WRITER word ONLY: 0 (free) or WRITER_BIT|pid. NOT a reader count. */
uint32_t rwait; /* 68 parked-waiter hint (readers+writers blocked on wlock); over-count-safe */
uint32_t drain_seq; /* 72 futex bumped by a reader releasing under a draining writer (wakes it) */
uint32_t slotless_rdepth; /* readers holding with no reader-slot (documented residual) */
uint64_t stat_ops; /* 80 */
uint8_t sealed; /* 88 0 = mutable, 1 = frozen (read-only; lock-free reads) */
uint8_t _pad[167]; /* 89..255 */
};
typedef struct ItHeader ItHeader;
_Static_assert(sizeof(ItHeader) == 256, "ItHeader must be 256 bytes");
/* One interval node: [lo, hi] with a user payload id, plus max_hi = the maximum
* hi endpoint over this node's whole subtree (the augmentation that lets a query
* prune subtrees that cannot reach the query point/range). A balanced BST keyed
* by lo; left/right are child indices (IT_NIL for none). */
typedef struct {
int64_t lo, hi;
int64_t max_hi;
uint64_t payload;
uint32_t left, right;
} ItNode;
_Static_assert(sizeof(ItNode) == 40, "ItNode must be 40 bytes");
/* ---- Process-local handle ---- */
typedef struct ItHandle {
ItHeader *hdr;
ItReaderSlot *reader_slots; /* IT_READER_SLOTS entries */
uint64_t *occ; /* IT_OCC_WORDS-word slot-occupancy bitmap (trusted layout offset) */
void *base; /* mmap base */
uint64_t nodes_off; /* validated offsets, cached: never re-read from the peer-writable header */
uint64_t idx_off;
uint32_t capacity; /* cached */
size_t mmap_size;
char *path; /* backing file path (strdup'd) */
int backing_fd; /* memfd or reopened-fd to close on destroy, -1 for file/anon */
uint32_t my_slot_idx; /* UINT32_MAX if all slots taken (no recovery for this handle) */
uint32_t cached_pid; /* getpid() cached at last slot claim */
uint32_t cached_fork_gen; /* it_fork_gen value at last slot claim */
uint32_t slotless_held; /* read-locks this process holds with no reader-slot */
int readonly; /* 1 = frozen O_RDONLY/PROT_READ view: lock-free reads, mutation croaks */
} ItHandle;
/* ================================================================
* Futex-based write-preferring read-write lock (reader-slots-only)
* with dead-process recovery
*
* The reader count is NOT stored in a shared counter. It is DISTRIBUTED across
* per-process reader slots: each slot's `rdepth` is that process's entire
* contribution to the lock. A reader publishes its presence in its own slot and
* then re-checks the writer word; a writer publishes the writer word and then
* scans every slot until all live readers' rdepth reach 0. Sequentially-
* consistent store+load on each side (a Dekker handshake) gives mutual exclusion.
*
* Because a reader's whole contribution is ONE atomic word owned by ONE process,
* a crashed reader is recovered by clearing that one slot (CAS its pid to 0) --
* there is no second counter to strand, no orphaned +1, and therefore no
* quiescent force-reset. A reader killed anywhere in rdlock/rdunlock leaves at
* most `rdepth>0` in its dead slot, which the draining writer clears directly, so
* sustained read traffic can never starve a writer. Write-preference is inherent
* in the gate (new readers see wlock!=0 and yield), so there is no reader-count
* yield hack.
* ================================================================ */
#define IT_RWLOCK_SPIN_LIMIT 32
#define IT_LOCK_TIMEOUT_SEC 2 /* FUTEX_WAIT timeout for stale-lock detection / drain re-scan */
static inline void it_rwlock_spin_pause(void) {
#if defined(__x86_64__) || defined(__i386__)
__asm__ volatile("pause" ::: "memory");
#elif defined(__aarch64__)
__asm__ volatile("yield" ::: "memory");
#else
intervaltree.h view on Meta::CPAN
}
ItHandle *h = it_setup(base, ms, path, -1); /* munmaps on OOM */
if (!h) { IT_ERR("out of memory"); return NULL; }
h->readonly = 1;
return h;
}
static void it_destroy(ItHandle *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(&it_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.) */
it_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 it_msync(ItHandle *h) {
if (!h || !h->base) return 0;
return msync(h->base, h->mmap_size, MS_SYNC);
}
/* forward decl: it_freeze (below) must force a pending build to completion
* before sealing; it_build_locked itself is defined further down, alongside
* the rest of the tree operations. */
static void it_build_locked(ItHandle *h);
/* Seal a tree: make it permanently immutable so it can be shipped and opened
* read-only. Takes the write lock and, if a build is pending (dirty), force-
* completes it right there -- a sealed tree must NEVER be left dirty, because
* the read-only query path (it_open_readonly handles) takes no lock and never
* builds; publishing the seal is what makes that lock-free path safe. Then
* flushes it (file/memfd-backed). Afterwards every mutator croaks and a
* read-write reopen is refused. */
static int it_freeze(ItHandle *h) {
it_rwlock_wrlock(h);
if (h->hdr->dirty) it_build_locked(h);
__atomic_signal_fence(__ATOMIC_SEQ_CST);
h->hdr->sealed = 1;
it_rwlock_wrunlock(h);
if (h->path || h->backing_fd >= 0) return it_msync(h);
return 0; /* anonymous: the seal lives in shared memory (visible to forks); nothing to flush */
}
/* ================================================================
* Interval-tree operations (callers hold the lock)
*
* Intervals are appended O(1) and marked dirty; a balanced BST keyed by the low
* endpoint is bulk-built on the first query after any insert (sorted once by lo,
* then median-split), so query recursion is O(log n) deep whatever the insertion
* order. Each node carries max_hi = the maximum hi endpoint over its whole
* subtree, so a query can prune a subtree whose intervals all end before it.
* Every child index read from shared memory is bounds-checked.
* ================================================================ */
/* one snapshotted matching interval, for a query result */
typedef struct { uint64_t id; int64_t lo, hi; } ItRes;
/* append an interval [lo,hi] with a payload id; returns its slot, or -1 if the
* tree is full. Marks the tree dirty so the next query rebuilds. (caller holds wrlock) */
static int64_t it_add_locked(ItHandle *h, int64_t lo, int64_t hi, uint64_t payload) {
uint64_t slot = h->hdr->count;
if (slot >= h->capacity) return -1; /* full */
ItNode *nd = it_node(h, slot);
nd->lo = lo; nd->hi = hi; nd->max_hi = hi; nd->payload = payload;
nd->left = IT_NIL; nd->right = IT_NIL;
/* Dirty before the count: a kill between costs a rebuild, never a counted node outside the tree. */
h->hdr->dirty = 1;
__atomic_signal_fence(__ATOMIC_SEQ_CST);
h->hdr->count = slot + 1;
return (int64_t)slot;
}
/* ---- balanced bulk build (sort once by lo, median split, augment max_hi) ---- */
static int it_cmp_lo(const void *pa, const void *pb, void *arg) {
ItHandle *h = (ItHandle *)arg;
uint32_t a = *(const uint32_t *)pa, b = *(const uint32_t *)pb;
int64_t la = it_node(h, a)->lo, lb = it_node(h, b)->lo;
if (la < lb) return -1;
if (la > lb) return 1;
return (a < b) ? -1 : (a > b ? 1 : 0); /* stable tiebreak */
}
/* build a balanced subtree over the already-lo-sorted idx[lo..hi]; returns the
* subtree root and sets its max_hi from its own hi and its children's max_hi. */
static uint32_t it_build_rec(ItHandle *h, uint32_t *idx, int64_t lo, int64_t hi) {
if (lo > hi) return IT_NIL;
int64_t mid = lo + (hi - lo) / 2;
uint32_t node = idx[mid];
uint32_t l = it_build_rec(h, idx, lo, mid - 1);
uint32_t r = it_build_rec(h, idx, mid + 1, hi);
ItNode *nd = it_node(h, node);
nd->left = l; nd->right = r;
int64_t mx = nd->hi;
if (l != IT_NIL && it_node(h, l)->max_hi > mx) mx = it_node(h, l)->max_hi;
if (r != IT_NIL && it_node(h, r)->max_hi > mx) mx = it_node(h, r)->max_hi;
nd->max_hi = mx;
return node;
}
/* (re)build a balanced tree over all inserted intervals (caller holds wrlock) */
static void it_build_locked(ItHandle *h) {
uint64_t n = h->hdr->count;
if (n > h->capacity) n = h->capacity; /* Layer B */
if (n == 0) { h->hdr->root = IT_NIL; __atomic_signal_fence(__ATOMIC_SEQ_CST); h->hdr->dirty = 0; return; }
uint32_t *idx = it_idx(h); /* scratch region inside the mapping */
for (uint64_t i = 0; i < n; i++) idx[i] = (uint32_t)i;
qsort_r(idx, (size_t)n, sizeof(uint32_t), it_cmp_lo, h); /* sort by lo once (subranges stay sorted) */
h->hdr->root = it_build_rec(h, idx, 0, (int64_t)n - 1);
__atomic_signal_fence(__ATOMIC_SEQ_CST);
h->hdr->dirty = 0;
( run in 1.470 second using v1.01-cache-2.11-cpan-16aabcbfbcb )