Data-Sync-Shared
view release on metacpan or search on metacpan
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);
__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;
}
}
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
}
/* 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) {
* 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))
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
}
/* 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;
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. */
* 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). */
* 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
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);
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;
}
( run in 4.525 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )