Data-Sync-Shared
view release on metacpan or search on metacpan
generation = increments each time barrier trips */
/* RWLock: value = WRITER word only (0=free, 0x80000000|pid=writer; readers
are NOT counted here -- each holds rdepth in its own reader-slot),
waiters = parked lockers (readers+writers) hint on the value futex */
/* Condvar: value = signal counter (futex word), waiters = blocked waiters,
mutex = associated mutex for predicate protection */
/* Once: value = state (0=INIT, 1=RUNNING|pid, 2=DONE),
waiters = blocked on completion */
uint32_t value; /* 64: primary state word (futex target) */
uint32_t waiters; /* 68: waiter count */
uint32_t generation; /* 72: barrier generation / condvar epoch */
uint32_t mutex; /* 76: condvar mutex (0 or PID|0x80000000) */
uint32_t mutex_waiters; /* 80: condvar mutex waiter count */
uint32_t stat_recoveries;/* 84 */
uint64_t stat_acquires; /* 88 */
uint64_t stat_releases; /* 96 */
uint64_t stat_waits; /* 104 */
uint64_t stat_timeouts; /* 112 */
uint32_t stat_signals; /* 120 */
uint32_t drain_seq; /* 124: RWLock futex a releasing reader bumps to wake a
writer draining readers in wrlock Phase 2 */
} SyncHeader;
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
_Static_assert(sizeof(SyncHeader) == 128, "SyncHeader must be 128 bytes");
#endif
/* ================================================================
* Process-local handle
* ================================================================ */
typedef struct {
SyncHeader *hdr;
size_t mmap_size;
char *path;
int notify_fd; /* eventfd, -1 if disabled */
int backing_fd; /* memfd fd, -1 for file-backed/anonymous */
SyncReaderSlot *reader_slots; /* in mmap, SYNC_READER_SLOTS entries; NULL if not RWLock */
uint64_t *occ; /* in mmap, SYNC_OCC_WORDS-word reader-slot occupancy bitmap; NULL if not RWLock (set alongside reader_slots) */
uint32_t my_slot_idx; /* UINT32_MAX = unclaimed; per-process slot index */
uint32_t cached_pid; /* getpid() at claim time */
uint32_t cached_fork_gen; /* fork-generation at claim time */
uint32_t slotless_held; /* rwlock read-locks this handle holds without a reader-slot */
} SyncHandle;
/* ================================================================
* Utility
* ================================================================ */
static inline void sync_spin_pause(void) {
#if defined(__x86_64__) || defined(__i386__)
__asm__ volatile("pause" ::: "memory");
#elif defined(__aarch64__)
__asm__ volatile("yield" ::: "memory");
#else
__asm__ volatile("" ::: "memory");
#endif
}
/* 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 sync_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 sync_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 !sync_pid_is_zombie(pid); /* kill() also succeeds for a zombie -> treat as dead */
}
/* Convert timeout in seconds (double) to absolute deadline */
static inline void sync_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;
}
}
/* Compute remaining timespec from absolute deadline. Returns 0 if deadline passed. */
static inline int sync_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;
}
/* ================================================================
* Mutex helpers (for Condvar's internal mutex)
* ================================================================ */
#define SYNC_MUTEX_WRITER_BIT 0x80000000U
#define SYNC_MUTEX_PID_MASK 0x7FFFFFFFU
#define SYNC_MUTEX_VAL(pid) (SYNC_MUTEX_WRITER_BIT | ((uint32_t)(pid) & SYNC_MUTEX_PID_MASK))
static const struct timespec sync_lock_timeout = { SYNC_LOCK_TIMEOUT_SEC, 0 };
static inline void sync_recover_stale_mutex(SyncHeader *hdr, uint32_t observed) {
if (!__atomic_compare_exchange_n(&hdr->mutex, &observed, 0,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED))
return;
__atomic_add_fetch(&hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->mutex_waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->mutex, FUTEX_WAKE, 1, NULL, NULL, 0);
}
static inline void sync_mutex_lock(SyncHeader *hdr) {
uint32_t mypid = SYNC_MUTEX_VAL((uint32_t)getpid());
for (int spin = 0; ; spin++) {
uint32_t expected = 0;
if (__atomic_compare_exchange_n(&hdr->mutex, &expected, mypid,
1, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED))
return;
if (__builtin_expect(spin < SYNC_SPIN_LIMIT, 1)) {
sync_spin_pause();
continue;
}
__atomic_add_fetch(&hdr->mutex_waiters, 1, __ATOMIC_RELAXED);
/* StoreLoad: publish mutex_waiters++ before re-reading mutex, so an
* unlocker sees our registration or we see the unlock (cur==0 -> retry). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
uint32_t cur = __atomic_load_n(&hdr->mutex, __ATOMIC_RELAXED);
if (cur != 0) {
long rc = syscall(SYS_futex, &hdr->mutex, FUTEX_WAIT, cur,
&sync_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
__atomic_sub_fetch(&hdr->mutex_waiters, 1, __ATOMIC_RELAXED);
uint32_t val = __atomic_load_n(&hdr->mutex, __ATOMIC_RELAXED);
if (val >= SYNC_MUTEX_WRITER_BIT) {
uint32_t pid = val & SYNC_MUTEX_PID_MASK;
if (!sync_pid_alive(pid))
sync_recover_stale_mutex(hdr, val);
}
spin = 0;
continue;
}
}
__atomic_sub_fetch(&hdr->mutex_waiters, 1, __ATOMIC_RELAXED);
spin = 0;
}
}
static inline void sync_mutex_unlock(SyncHeader *hdr) {
__atomic_store_n(&hdr->mutex, 0, __ATOMIC_RELEASE);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->mutex_waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->mutex, FUTEX_WAKE, 1, NULL, NULL, 0);
}
/* ================================================================
* Futex-based write-preferring read-write lock (reader-slots-only)
* for SYNC_TYPE_RWLOCK, with dead-process recovery
*
* value == 0: unlocked (no writer)
* value == 0x80000000 | pid: write-locked by pid
*
* The reader count is NOT stored in `value`. 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 `value`; a writer publishes `value` 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) --
* no second counter to strand, no orphaned +1, 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 value!=0 and yield).
* ================================================================ */
#define SYNC_RWLOCK_WRITER_BIT 0x80000000U
#define SYNC_RWLOCK_PID_MASK 0x7FFFFFFFU
#define SYNC_RWLOCK_WR(pid) (SYNC_RWLOCK_WRITER_BIT | ((uint32_t)(pid) & SYNC_RWLOCK_PID_MASK))
static inline int sync_rwlock_try_rdlock(SyncHandle *h);
static inline int sync_rwlock_try_wrlock(SyncHandle *h);
/* Force-recover a stale WRITE lock left by a dead writer (held or mid-drain).
* A single CAS observed->0 wins the recovery race (a loser's CAS just fails);
* this module has no extra shared lock state to repair. Bumps stat_recoveries
* and wakes any lockers parked on the value futex. */
static inline void sync_recover_stale_rwlock(SyncHeader *hdr, uint32_t observed) {
if (!__atomic_compare_exchange_n(&hdr->value, &observed, 0,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED))
return;
__atomic_add_fetch(&hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
* Each process claims one SyncReaderSlot lazily on first rwlock op so that a
* SIGKILL'd reader's rdepth contribution can be neutralised by a draining
* writer. Only relevant for SYNC_TYPE_RWLOCK; non-RWLock primitives leave
* h->reader_slots == NULL and this helper is a no-op. */
static uint32_t sync_fork_gen = 0;
static pthread_once_t sync_atfork_once = PTHREAD_ONCE_INIT;
static void sync_on_fork_child(void) {
__atomic_add_fetch(&sync_fork_gen, 1, __ATOMIC_RELAXED);
}
static void sync_atfork_init(void) {
pthread_atfork(NULL, NULL, sync_on_fork_child);
}
/* Occupancy bitmap: set a slot's bit when it is claimed, clear it on clean
* release. SEQ_CST so a set bit is ordered before that slot's rdepth can go
* non-zero (the bit is set in claim, which precedes any rdlock's rdepth++),
* letting a writer's SEQ_CST bitmap scan never miss a slot a committed reader
* holds. Callers guarantee h->occ != NULL (it is set alongside h->reader_slots,
* and every call site is under an `if (h->reader_slots)` guard). */
static inline void sync_occ_set(SyncHandle *h, uint32_t s) {
__atomic_fetch_or(&h->occ[s >> 6], (uint64_t)1 << (s & 63), __ATOMIC_SEQ_CST);
}
static inline void sync_occ_clear(SyncHandle *h, uint32_t s) {
__atomic_fetch_and(&h->occ[s >> 6], ~((uint64_t)1 << (s & 63)), __ATOMIC_SEQ_CST);
}
static inline void sync_claim_reader_slot(SyncHandle *h) {
if (!h->reader_slots) return;
pthread_once(&sync_atfork_once, sync_atfork_init);
uint32_t cur_gen = __atomic_load_n(&sync_fork_gen, __ATOMIC_RELAXED);
if (h->cached_fork_gen != cur_gen) {
h->slotless_held = 0; /* fork: child holds none of the parent's slotless read locks */
h->cached_fork_gen = cur_gen;
h->my_slot_idx = UINT32_MAX;
}
if (h->my_slot_idx != UINT32_MAX) return;
uint32_t now_pid = (uint32_t)getpid();
h->cached_pid = now_pid;
uint32_t start = now_pid % SYNC_READER_SLOTS;
/* Pass 1: take a free slot. */
for (uint32_t i = 0; i < SYNC_READER_SLOTS; i++) {
uint32_t s = (start + i) % SYNC_READER_SLOTS;
uint32_t expected = 0;
if (__atomic_compare_exchange_n(&h->reader_slots[s].pid,
&expected, now_pid, 0,
__ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
/* Fresh owner holds no read locks yet; clear any stale rdepth left by
* a dead predecessor (its contribution is dropped as we take over). */
__atomic_store_n(&h->reader_slots[s].rdepth, 0, __ATOMIC_RELAXED);
sync_occ_set(h, s); /* mark occupied BEFORE any rdlock can bump rdepth */
h->my_slot_idx = s;
return;
}
}
/* Pass 2: no free slot -- reclaim one whose owner is dead. Safe to take even
* if its rdepth>0: clearing pid drops the dead reader's entire contribution
* (a writer scan ignores rdepth when pid==0) and we reset rdepth to 0 as we
* claim it. */
for (uint32_t i = 0; i < SYNC_READER_SLOTS; i++) {
uint32_t dpid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (dpid == 0 || dpid == now_pid || sync_pid_alive(dpid)) continue;
uint32_t expected = dpid;
if (__atomic_compare_exchange_n(&h->reader_slots[i].pid, &expected, now_pid, 0,
__ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
__atomic_store_n(&h->reader_slots[i].rdepth, 0, __ATOMIC_RELAXED);
sync_occ_set(h, i);
h->my_slot_idx = i;
return;
}
}
/* Table full -- leave my_slot_idx = UINT32_MAX so this handle takes the
* slotless path (lock still works; recovery of THIS reader's death is the
* documented slotless limitation). */
}
/* Inspect the writer word after a futex-wait timeout. If a dead writer holds
* it, force-recover. Dead READERS need no action here: only a writer that owns
* `value` drains readers, and it clears dead readers inline in its own scan. */
static inline void sync_recover_after_timeout(SyncHandle *h) {
SyncHeader *hdr = h->hdr;
uint32_t val = __atomic_load_n(&hdr->value, __ATOMIC_RELAXED);
if (val >= SYNC_RWLOCK_WRITER_BIT) {
uint32_t pid = val & SYNC_RWLOCK_PID_MASK;
if (!sync_pid_alive(pid))
sync_recover_stale_rwlock(hdr, val);
}
}
/* Bump/drop the parked-waiter hint. Both readers (blocked at the gate) and
* writers (blocked acquiring `value`) wait on the value futex and use this, so
* wrunlock/recover know whether a FUTEX_WAKE is worth a syscall. A waiter
* SIGKILLed while parked leaves waiters over-counted -> at most a spurious wake
* (harmless); it can never under-count, so no wakeup is lost. */
static inline void sync_park(SyncHandle *h) {
__atomic_add_fetch(&h->hdr->waiters, 1, __ATOMIC_RELAXED);
}
static inline void sync_unpark(SyncHandle *h) {
__atomic_sub_fetch(&h->hdr->waiters, 1, __ATOMIC_RELAXED);
}
/* Publish (inc) / retract (dec) this reader's presence -- its ENTIRE
* contribution to the lock. A slotted reader uses its slot's rdepth; a reader
* that could not claim a slot uses the global slotless_rdepth. inc() is SEQ_CST
* so the value re-check that follows it in rdlock forms a Dekker handshake with
* the writer's SEQ_CST value-store + rdepth-scan. dec() peels slotless first so
* a slot claimed mid-hold cannot misattribute the decrement. */
static inline void sync_rdepth_inc(SyncHandle *h) {
if (h->my_slot_idx != UINT32_MAX) {
__atomic_add_fetch(&h->reader_slots[h->my_slot_idx].rdepth, 1, __ATOMIC_SEQ_CST);
} else {
__atomic_add_fetch(&h->hdr->slotless_rdepth, 1, __ATOMIC_SEQ_CST);
h->slotless_held++;
}
}
static inline void sync_rdepth_dec(SyncHandle *h) {
if (h->slotless_held > 0) {
h->slotless_held--;
__atomic_sub_fetch(&h->hdr->slotless_rdepth, 1, __ATOMIC_RELEASE);
} else if (h->my_slot_idx != UINT32_MAX) {
__atomic_sub_fetch(&h->reader_slots[h->my_slot_idx].rdepth, 1, __ATOMIC_RELEASE);
}
}
/* Wake a writer that may be draining readers (it waits on drain_seq). Called
* after every rdepth decrement so a released read lock lets the writer re-scan
* promptly instead of waiting out its timeout. */
static inline void sync_reader_wake_drain(SyncHandle *h) {
if (__atomic_load_n(&h->hdr->value, __ATOMIC_ACQUIRE) != 0) {
__atomic_add_fetch(&h->hdr->drain_seq, 1, __ATOMIC_RELEASE);
syscall(SYS_futex, &h->hdr->drain_seq, FUTEX_WAKE, 1, NULL, NULL, 0);
}
}
/* Give-up path for the timed/try readers: retract our published rdepth and wake
* a draining writer so it re-scans without our contribution. */
static inline void sync_reader_abort(SyncHandle *h) {
sync_rdepth_dec(h);
sync_reader_wake_drain(h);
}
/* Writer-side scan of all reader slots: reclaim any dead reader (CAS its pid to
* 0) and report whether a LIVE reader still holds (rdepth>0), including the
* slotless residual. Shared by wrlock Phase 2, try_wrlock, and wrlock_timed. */
static inline int sync_rwlock_readers_busy(SyncHandle *h) {
int busy = 0;
if (h->reader_slots) {
/* Visit only OCCUPIED slots via the occupancy bitmap (SEQ_CST: a committed
* reader's bit -- set in claim, before its rdepth++ -- is ordered before
* this scan, so no held slot is skipped). O(SYNC_OCC_WORDS + live readers)
* instead of O(SYNC_READER_SLOTS). */
for (uint32_t w = 0; w < SYNC_OCC_WORDS; w++) {
uint64_t word = __atomic_load_n(&h->occ[w], __ATOMIC_SEQ_CST);
while (word) {
uint32_t i = (w << 6) + (uint32_t)__builtin_ctzll(word);
word &= word - 1; /* consume this bit (local copy) */
uint32_t rd = __atomic_load_n(&h->reader_slots[i].rdepth, __ATOMIC_SEQ_CST);
if (rd == 0) continue; /* occupied but not read-locking now */
uint32_t pid = __atomic_load_n(&h->reader_slots[i].pid, __ATOMIC_ACQUIRE);
if (pid == 0) continue; /* no live owner -- stale rdepth, ignore */
if (!sync_pid_alive(pid)) {
/* Dead reader: clear its pid so the slot no longer counts (its
* whole contribution WAS this slot). Leave the occ bit SET
* (harmless -- a later scan hits pid==0 and skips, a re-claim
* re-sets it) to avoid racing a concurrent claimant. rdepth is
* left stale but is now ignored (pid==0) and zeroed by the next
* claimant. Count it as a recovery (gated on the CAS so it is
* tallied exactly once). */
uint32_t ep = pid;
if (__atomic_compare_exchange_n(&h->reader_slots[i].pid, &ep, 0,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED))
__atomic_add_fetch(&h->hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
continue;
}
busy = 1; /* live reader still holding */
}
}
}
/* A live slotless reader keeps us waiting; a crashed slotless reader that
* cannot be attributed to a pid is the documented slotless limitation. */
if (__atomic_load_n(&h->hdr->slotless_rdepth, __ATOMIC_SEQ_CST) != 0)
busy = 1;
return busy;
}
static inline void sync_rwlock_rdlock(SyncHandle *h) {
sync_claim_reader_slot(h);
SyncHeader *hdr = h->hdr;
for (int spin = 0; ; spin++) {
uint32_t cur = __atomic_load_n(&hdr->value, __ATOMIC_ACQUIRE);
if (cur == 0) {
/* Optimistically take the read: publish rdepth, then re-check value.
* SEQ_CST inc + SEQ_CST load vs the writer's SEQ_CST value CAS +
* SEQ_CST rdepth scan: by the single total order of SEQ_CST ops the
* two sides cannot both miss each other, so we never hold
* concurrently with a writer. */
sync_rdepth_inc(h);
if (__atomic_load_n(&hdr->value, __ATOMIC_SEQ_CST) == 0)
return; /* no writer after our publish -> we hold */
/* A writer appeared during our publish -- yield to it (write-preferring). */
sync_reader_abort(h); /* retract rdepth + wake the draining writer */
spin = 0;
continue;
}
/* value != 0: a writer holds or is acquiring. Recover if it is dead. */
if (cur >= SYNC_RWLOCK_WRITER_BIT &&
!sync_pid_alive(cur & SYNC_RWLOCK_PID_MASK)) {
sync_recover_stale_rwlock(hdr, cur);
spin = 0;
continue;
}
if (__builtin_expect(spin < SYNC_SPIN_LIMIT, 1)) {
sync_spin_pause();
continue;
}
sync_park(h);
/* StoreLoad: publish our parked registration before re-reading value
* (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
cur = __atomic_load_n(&hdr->value, __ATOMIC_RELAXED);
if (cur != 0) {
long rc = syscall(SYS_futex, &hdr->value, FUTEX_WAIT, cur,
&sync_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
sync_unpark(h);
sync_recover_after_timeout(h);
spin = 0;
continue;
}
}
sync_unpark(h);
spin = 0;
}
}
static inline void sync_rwlock_rdunlock(SyncHandle *h) {
sync_rdepth_dec(h); /* RELEASE: drop our entire contribution */
sync_reader_wake_drain(h); /* if a writer is draining, wake it to re-scan */
}
/* try_rdlock: one-shot, non-blocking. Returns 1 on success, 0 on failure. */
static inline int sync_rwlock_try_rdlock(SyncHandle *h) {
sync_claim_reader_slot(h);
SyncHeader *hdr = h->hdr;
uint32_t cur = __atomic_load_n(&hdr->value, __ATOMIC_ACQUIRE);
if (cur != 0) {
/* A writer holds or is acquiring. Recover once if it is dead, re-read. */
if (cur >= SYNC_RWLOCK_WRITER_BIT &&
!sync_pid_alive(cur & SYNC_RWLOCK_PID_MASK)) {
sync_recover_stale_rwlock(hdr, cur);
cur = __atomic_load_n(&hdr->value, __ATOMIC_ACQUIRE);
}
if (cur != 0) return 0;
}
sync_rdepth_inc(h);
if (__atomic_load_n(&hdr->value, __ATOMIC_SEQ_CST) != 0) {
/* A writer appeared during our publish -- yield (write-preferring). */
sync_reader_abort(h);
return 0;
}
return 1;
}
/* Timed rdlock: returns 1 on success, 0 on timeout. timeout<0 = infinite,
* timeout==0 = one-shot try. Same optimistic gate as sync_rwlock_rdlock; the
* gate FUTEX_WAIT is bounded by an absolute deadline computed from `timeout`.
* On give-up we hold NOTHING (every 0-return has already retracted rdepth). */
static inline int sync_rwlock_rdlock_timed(SyncHandle *h, double timeout) {
if (timeout == 0) return sync_rwlock_try_rdlock(h);
sync_claim_reader_slot(h);
SyncHeader *hdr = h->hdr;
struct timespec deadline, remaining;
int has_deadline = (timeout > 0);
if (has_deadline) sync_make_deadline(timeout, &deadline);
for (int spin = 0; ; spin++) {
uint32_t cur = __atomic_load_n(&hdr->value, __ATOMIC_ACQUIRE);
if (cur == 0) {
/* Optimistic publish + Dekker re-check (see sync_rwlock_rdlock). */
sync_rdepth_inc(h);
if (__atomic_load_n(&hdr->value, __ATOMIC_SEQ_CST) == 0)
return 1;
sync_reader_abort(h); /* retract rdepth + wake the draining writer */
if (has_deadline && !sync_remaining_time(&deadline, &remaining))
return 0; /* deadline passed; we hold nothing */
spin = 0;
continue;
}
if (cur >= SYNC_RWLOCK_WRITER_BIT &&
!sync_pid_alive(cur & SYNC_RWLOCK_PID_MASK)) {
sync_recover_stale_rwlock(hdr, cur);
spin = 0;
continue;
}
if (__builtin_expect(spin < SYNC_SPIN_LIMIT, 1)) {
sync_spin_pause();
continue;
}
sync_park(h);
/* StoreLoad: publish our parked registration before re-reading value. */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
cur = __atomic_load_n(&hdr->value, __ATOMIC_RELAXED);
if (cur != 0) {
/* Cap the wait at SYNC_LOCK_TIMEOUT_SEC so stale-holder recovery runs
* periodically even with a longer user deadline. */
struct timespec *pts = (struct timespec *)&sync_lock_timeout;
int capped = 1;
if (has_deadline) {
if (!sync_remaining_time(&deadline, &remaining)) {
sync_unpark(h);
return 0; /* deadline passed; nothing held */
}
if (remaining.tv_sec < SYNC_LOCK_TIMEOUT_SEC) { pts = &remaining; capped = 0; }
}
long rc = syscall(SYS_futex, &hdr->value, FUTEX_WAIT, cur, pts, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
sync_unpark(h);
if (!capped) return 0; /* user deadline expired; nothing held */
sync_recover_after_timeout(h);
spin = 0;
continue;
}
}
sync_unpark(h);
spin = 0;
}
}
static inline void sync_rwlock_wrlock(SyncHandle *h) {
sync_claim_reader_slot(h); /* refresh cached_pid + own a slot across fork */
SyncHeader *hdr = h->hdr;
/* Encode PID in the value word itself (0x80000000 | pid) to eliminate any
* crash window between acquiring the lock and storing the owner. */
uint32_t mypid = SYNC_RWLOCK_WR((uint32_t)getpid());
/* Phase 1: acquire the writer word (mutual exclusion among writers). */
for (int spin = 0; ; spin++) {
uint32_t expected = 0;
if (__atomic_compare_exchange_n(&hdr->value, &expected, mypid,
0, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED))
break;
/* Contended: expected now holds the current value. Recover a dead writer. */
if (expected >= SYNC_RWLOCK_WRITER_BIT &&
!sync_pid_alive(expected & SYNC_RWLOCK_PID_MASK)) {
sync_recover_stale_rwlock(hdr, expected);
spin = 0;
continue;
}
if (__builtin_expect(spin < SYNC_SPIN_LIMIT, 1)) {
sync_spin_pause();
continue;
}
sync_park(h);
/* StoreLoad: see the rdlock park path (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
uint32_t cur = __atomic_load_n(&hdr->value, __ATOMIC_RELAXED);
if (cur != 0) {
long rc = syscall(SYS_futex, &hdr->value, FUTEX_WAIT, cur,
&sync_lock_timeout, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
sync_unpark(h);
sync_recover_after_timeout(h);
spin = 0;
continue;
}
}
sync_unpark(h);
spin = 0;
}
/* Phase 2: we own `value`, so no NEW reader can join (they see value!=0 and
* yield). Drain the readers that were already holding when we won the CAS.
* The SEQ_CST CAS above + the SEQ_CST rdepth loads below are the writer side
* of the Dekker handshake. */
for (;;) {
uint32_t v = __atomic_load_n(&hdr->drain_seq, __ATOMIC_RELAXED); /* snapshot BEFORE scan */
if (!sync_rwlock_readers_busy(h))
return; /* exclusive: value held + every rdepth 0 */
/* Wait for a reader to release (drain_seq bump) or time out to re-scan
* (which reclaims any newly-dead slotted reader). */
syscall(SYS_futex, &hdr->drain_seq, FUTEX_WAIT, v, &sync_lock_timeout, NULL, 0);
}
}
static inline void sync_rwlock_wrunlock(SyncHandle *h) {
SyncHeader *hdr = h->hdr;
__atomic_store_n(&hdr->value, 0, __ATOMIC_RELEASE);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->value, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
/* try_wrlock: one-shot. Returns 1 iff we obtain exclusive ownership, 0
* otherwise. A try must NOT wait for readers to drain. */
static inline int sync_rwlock_try_wrlock(SyncHandle *h) {
sync_claim_reader_slot(h);
SyncHeader *hdr = h->hdr;
uint32_t mypid = SYNC_RWLOCK_WR((uint32_t)getpid());
uint32_t expected = 0;
if (!__atomic_compare_exchange_n(&hdr->value, &expected, mypid,
0, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED)) {
/* Contended. Recover a dead writer and retry once; else fail. */
if (expected >= SYNC_RWLOCK_WRITER_BIT &&
!sync_pid_alive(expected & SYNC_RWLOCK_PID_MASK)) {
sync_recover_stale_rwlock(hdr, expected);
expected = 0;
if (!__atomic_compare_exchange_n(&hdr->value, &expected, mypid,
0, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED))
return 0;
} else {
return 0;
}
}
/* We hold `value`. Dekker: our SEQ_CST CAS + the SEQ_CST rdepth loads in the
* scan cannot both miss a reader whose SEQ_CST publish + value re-check ran. */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (sync_rwlock_readers_busy(h)) {
/* Readers still hold -- release the writer word and fail (no waiting). */
__atomic_store_n(&hdr->value, 0, __ATOMIC_RELEASE);
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->value, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
return 0;
}
return 1; /* exclusive: value held + every rdepth 0 */
}
/* Timed wrlock: returns 1 on success, 0 on timeout. timeout<0 = infinite,
* timeout==0 = one-shot try. BOTH the Phase-1 value acquire and the Phase-2
* drain wait are bounded by an absolute deadline; on a Phase-2 give-up we own
* `value` and MUST release it before returning 0. */
static inline int sync_rwlock_wrlock_timed(SyncHandle *h, double timeout) {
if (timeout == 0) return sync_rwlock_try_wrlock(h);
sync_claim_reader_slot(h);
SyncHeader *hdr = h->hdr;
uint32_t mypid = SYNC_RWLOCK_WR((uint32_t)getpid());
struct timespec deadline, remaining;
int has_deadline = (timeout > 0);
if (has_deadline) sync_make_deadline(timeout, &deadline);
/* Phase 1: acquire the writer word, bounded by the deadline. */
for (int spin = 0; ; spin++) {
uint32_t expected = 0;
if (__atomic_compare_exchange_n(&hdr->value, &expected, mypid,
0, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED))
break;
if (expected >= SYNC_RWLOCK_WRITER_BIT &&
!sync_pid_alive(expected & SYNC_RWLOCK_PID_MASK)) {
sync_recover_stale_rwlock(hdr, expected);
spin = 0;
continue;
}
if (__builtin_expect(spin < SYNC_SPIN_LIMIT, 1)) {
sync_spin_pause();
continue;
}
sync_park(h);
__atomic_thread_fence(__ATOMIC_SEQ_CST);
uint32_t cur = __atomic_load_n(&hdr->value, __ATOMIC_RELAXED);
if (cur != 0) {
struct timespec *pts = (struct timespec *)&sync_lock_timeout;
int capped = 1;
if (has_deadline) {
if (!sync_remaining_time(&deadline, &remaining)) {
sync_unpark(h);
return 0; /* deadline passed; no lock held */
}
if (remaining.tv_sec < SYNC_LOCK_TIMEOUT_SEC) { pts = &remaining; capped = 0; }
}
long rc = syscall(SYS_futex, &hdr->value, FUTEX_WAIT, cur, pts, NULL, 0);
if (rc == -1 && errno == ETIMEDOUT) {
sync_unpark(h);
if (!capped) return 0; /* user deadline expired; no lock held */
sync_recover_after_timeout(h);
spin = 0;
continue;
}
}
sync_unpark(h);
spin = 0;
}
/* Phase 2: we own `value`; drain existing readers, bounded by the deadline.
* On a deadline give-up we must release the writer word we hold. */
for (;;) {
uint32_t v = __atomic_load_n(&hdr->drain_seq, __ATOMIC_RELAXED);
if (!sync_rwlock_readers_busy(h))
return 1; /* exclusive */
struct timespec *pts = (struct timespec *)&sync_lock_timeout;
if (has_deadline) {
if (!sync_remaining_time(&deadline, &remaining)) {
/* Deadline hit while draining -- release the value word we own. */
__atomic_store_n(&hdr->value, 0, __ATOMIC_RELEASE);
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->value, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
return 0;
}
if (remaining.tv_sec < SYNC_LOCK_TIMEOUT_SEC) pts = &remaining;
}
syscall(SYS_futex, &hdr->drain_seq, FUTEX_WAIT, v, pts, NULL, 0);
}
}
/* Downgrade: convert a held write lock (value == WRITER|pid) to a read lock.
* Publish our reader rdepth FIRST (SEQ_CST) so a writer that later CASes `value`
* sees our rdepth in its drain scan, THEN release the writer word. */
static inline void sync_rwlock_downgrade(SyncHandle *h) {
sync_claim_reader_slot(h);
*
* value states: 0=INIT, (SYNC_MUTEX_WRITER_BIT|pid)=RUNNING, 1=DONE
* ================================================================ */
#define SYNC_ONCE_INIT 0
#define SYNC_ONCE_DONE 1
/* RUNNING = SYNC_MUTEX_WRITER_BIT | pid */
static inline int sync_once_is_done(SyncHandle *h) {
return __atomic_load_n(&h->hdr->value, __ATOMIC_ACQUIRE) == SYNC_ONCE_DONE;
}
/* Try to become the initializer. Returns:
* 1 = you are the initializer, call once_done() when finished
* 0 = already done
* -1 = another process is initializing (wait with once_wait) */
static inline int sync_once_try(SyncHandle *h) {
SyncHeader *hdr = h->hdr;
uint32_t mypid = SYNC_MUTEX_VAL((uint32_t)getpid());
uint32_t expected = SYNC_ONCE_INIT;
if (__atomic_compare_exchange_n(&hdr->value, &expected, mypid,
0, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
__atomic_add_fetch(&hdr->stat_acquires, 1, __ATOMIC_RELAXED);
return 1;
}
if (expected == SYNC_ONCE_DONE) return 0;
return -1;
}
/* Call/wait combo: try to become initializer, or wait for completion.
* Returns 1 if caller is the initializer, 0 if already done or waited. */
static inline int sync_once_enter(SyncHandle *h, double timeout) {
SyncHeader *hdr = h->hdr;
/* Non-blocking probe: just try, don't wait */
int r = sync_once_try(h);
if (r == 1) return 1;
if (r == 0) return 0;
if (timeout == 0) return 0;
struct timespec deadline, remaining;
int has_deadline = (timeout > 0);
if (has_deadline) sync_make_deadline(timeout, &deadline);
__atomic_add_fetch(&hdr->stat_waits, 1, __ATOMIC_RELAXED);
for (;;) {
r = sync_once_try(h);
if (r == 1) return 1; /* caller is initializer */
if (r == 0) return 0; /* already done */
/* r == -1: someone else is running. Wait or detect stale. */
uint32_t val = __atomic_load_n(&hdr->value, __ATOMIC_ACQUIRE);
if (val == SYNC_ONCE_DONE) return 0;
if (val == SYNC_ONCE_INIT) continue; /* race: was reset, retry */
/* Check stale initializer */
if (val >= SYNC_MUTEX_WRITER_BIT) {
uint32_t pid = val & SYNC_MUTEX_PID_MASK;
if (!sync_pid_alive(pid)) {
if (__atomic_compare_exchange_n(&hdr->value, &val, SYNC_ONCE_INIT,
0, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED)) {
__atomic_add_fetch(&hdr->stat_recoveries, 1, __ATOMIC_RELAXED);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->value, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
continue;
}
}
__atomic_add_fetch(&hdr->waiters, 1, __ATOMIC_RELAXED);
/* Always cap at SYNC_LOCK_TIMEOUT_SEC so stale-initializer recovery
* runs periodically even when the caller specifies infinite timeout. */
struct timespec *pts = (struct timespec *)&sync_lock_timeout;
if (has_deadline) {
if (!sync_remaining_time(&deadline, &remaining)) {
__atomic_sub_fetch(&hdr->waiters, 1, __ATOMIC_RELAXED);
__atomic_add_fetch(&hdr->stat_timeouts, 1, __ATOMIC_RELAXED);
return 0;
}
if (remaining.tv_sec < SYNC_LOCK_TIMEOUT_SEC)
pts = &remaining;
}
syscall(SYS_futex, &hdr->value, FUTEX_WAIT, val, pts, NULL, 0);
__atomic_sub_fetch(&hdr->waiters, 1, __ATOMIC_RELAXED);
}
}
static inline void sync_once_done(SyncHandle *h) {
SyncHeader *hdr = h->hdr;
__atomic_store_n(&hdr->value, SYNC_ONCE_DONE, __ATOMIC_RELEASE);
__atomic_add_fetch(&hdr->stat_releases, 1, __ATOMIC_RELAXED);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->value, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
static inline void sync_once_reset(SyncHandle *h) {
SyncHeader *hdr = h->hdr;
__atomic_store_n(&hdr->value, SYNC_ONCE_INIT, __ATOMIC_RELEASE);
/* StoreLoad: publish the state change before reading the waiter count (weak-memory lost-wakeup guard). */
__atomic_thread_fence(__ATOMIC_SEQ_CST);
if (__atomic_load_n(&hdr->waiters, __ATOMIC_RELAXED) > 0)
syscall(SYS_futex, &hdr->value, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}
/* ================================================================
* Create / Open / Close
*
* Layout:
* [0..127] : SyncHeader
* [128 .. 128+SLOTS_SIZE-1] : SyncReaderSlot[SYNC_READER_SLOTS] (RWLock only)
* [128+SLOTS_SIZE .. +OCC_BYTES-1] : occupancy bitmap, SYNC_OCC_WORDS words (RWLock only)
*
* Non-RWLock primitives keep total_size = sizeof(SyncHeader) (Option A:
( run in 0.551 second using v1.01-cache-2.11-cpan-14f38c9f855 )