/ 
Data-Sync-Shared-0.07
/ Data::Sync::Shared

NAME

Data::Sync::Shared - Shared-memory synchronization primitives for Linux

SYNOPSIS

use Data::Sync::Shared;

# Semaphore -- bounded counter for resource limiting
my $sem = Data::Sync::Shared::Semaphore->new('/tmp/sem.shm', 4);
$sem->acquire;            # block until available
$sem->acquire(1.5);       # with timeout
$sem->try_acquire;        # non-blocking
$sem->acquire_n(3);       # acquire N permits atomically
$sem->release;
$sem->release(2);         # release N permits
my $n = $sem->drain;      # acquire all, return count
{
    my $g = $sem->acquire_guard;   # auto-release on scope exit
}

# Barrier -- N processes rendezvous
my $bar = Data::Sync::Shared::Barrier->new('/tmp/bar.shm', 3);
my $leader = $bar->wait;       # block until all 3 arrive
my $leader = $bar->wait(5.0);  # with timeout (-1=timeout)

# RWLock -- reader-writer lock
my $rw = Data::Sync::Shared::RWLock->new('/tmp/rw.shm');
$rw->rdlock;  $rw->rdunlock;
$rw->wrlock;  $rw->wrunlock;
$rw->try_rdlock;  $rw->try_wrlock;
$rw->downgrade;                # wrlock -> rdlock atomically
{
    my $g = $rw->wrlock_guard;  # auto-release on scope exit
}

# Condvar -- condition variable with built-in mutex
my $cv = Data::Sync::Shared::Condvar->new('/tmp/cv.shm');
$cv->lock;
$cv->try_lock;       # non-blocking
$cv->wait;           # atomically unlock + wait + re-lock
$cv->wait(2.0);      # with timeout
$cv->signal;         # wake one waiter
$cv->broadcast;      # wake all waiters
$cv->wait_while(sub { !$ready }, 5.0);  # predicate loop
$cv->unlock;

# Once -- one-time initialization gate
my $once = Data::Sync::Shared::Once->new('/tmp/once.shm');
if ($once->enter) {          # or enter($timeout)
    do_init();
    $once->done;
}

# All primitives support anonymous (fork-inherited) mode:
my $sem = Data::Sync::Shared::Semaphore->new(undef, 4);

# And memfd mode (fd-passable):
my $sem = Data::Sync::Shared::Semaphore->new_memfd("my_sem", 4);
my $fd = $sem->memfd;

DESCRIPTION

Data::Sync::Shared provides five cross-process synchronization primitives stored in file-backed shared memory (mmap(MAP_SHARED)), using Linux futex for efficient blocking.

Linux-only. Requires 64-bit Perl.

Primitives

Data::Sync::Shared::Semaphore - bounded counter

CAS-based counting semaphore. acquire decrements (blocks at 0), release increments (capped at max). Useful for cross-process resource limiting (connection pools, worker slots).

Data::Sync::Shared::Barrier - rendezvous point

N processes call wait; all block until the last one arrives, then all proceed. Returns true for one "leader" process. Generation counter tracks how many times the barrier has tripped.

Data::Sync::Shared::RWLock - reader-writer lock

Multiple concurrent readers or one exclusive writer. Readers use rdlock/rdunlock, writers use wrlock/wrunlock. Non-blocking try_rdlock/try_wrlock variants available.

Data::Sync::Shared::Condvar - condition variable

Includes a built-in mutex. lock/unlock protect the predicate. wait atomically releases the mutex and sleeps; on wakeup it re-acquires the mutex. signal wakes one waiter, broadcast wakes all.

Data::Sync::Shared::Once - one-time init gate

enter returns true for exactly one process (the initializer); all others block until done is called. If the initializer dies, waiters detect the stale PID and a new initializer is elected.

Features

  • File-backed mmap for cross-process sharing

  • Futex-based blocking (no busy-spin, no pthread)

  • PID-based stale lock recovery (dead process detection)

  • Anonymous and memfd modes

  • Timeouts on all blocking operations (except Condvar lock, which provides only a non-blocking try_lock)

  • eventfd integration for event-loop wakeup

Crash Safety

All primitives encode the holder's PID in the lock word. If a process dies while holding a lock, other processes detect the stale lock within 2 seconds via kill(pid, 0) and automatically recover.

Reader-slot exhaustion (slotless readers): dead-process recovery attributes a crashed lock holder's contribution through its reader-slot. The slot table holds 1024 entries (one per concurrent reader process). If more than that many reader processes share one mapping at once, a reader that cannot claim a slot proceeds "slotless" -- it still takes the read lock but leaves no per-process record. If such a slotless reader is then killed while holding the read lock, its share of the lock cannot be attributed to a dead process, so writer recovery cannot reclaim it and writers may block until the mapping is recreated. Reaching this needs more than 1024 concurrent reader processes on one mapping plus a crash in the brief read-lock window; the dead-process slot reclaim keeps the table from filling with stale entries, so in practice it is very unlikely.

Security

Backing files are created securely: the path is opened with O_CREAT|O_EXCL|O_NOFOLLOW (a pre-existing symlink at the path is rejected, and an existing regular file is attached rather than truncated) and the new file is created mode 0600 (owner-only) by default, so a segment is not world-writable unless you opt in. To share a file with a peer group, pass an explicit octal mode as the last argument to new():

my $sem = Data::Sync::Shared::Semaphore->new($path, $max, $initial, 0660);
my $rw  = Data::Sync::Shared::RWLock->new($path, 0660);

The mode is applied exactly via fchmod after an O_EXCL create, so the caller's umask does not narrow it. Offsets read back from an attached segment are bounds-checked before use, so a poisoned reader-slot offset cannot steer a pointer outside the mapping.

Guard Objects

All locking primitives provide scope-based guards that auto-release on scope exit (including exceptions):

{
    my $g = $rw->rdlock_guard;
    # ... read operations ...
}  # rdunlock called automatically

{
    my $g = $sem->acquire_guard(3);  # acquire 3 permits
    # ... use resource ...
}  # release(3) called automatically

{
    my $g = $cv->lock_guard;
    $cv->wait_while(sub { !$ready }, 5.0);
}  # unlock called automatically

PRIMITIVES

Data::Sync::Shared::Semaphore

Constructors

my $sem = Data::Sync::Shared::Semaphore->new($path, $max);
my $sem = Data::Sync::Shared::Semaphore->new($path, $max, $initial);
my $sem = Data::Sync::Shared::Semaphore->new(undef, $max);
my $sem = Data::Sync::Shared::Semaphore->new_memfd($name, $max);
my $sem = Data::Sync::Shared::Semaphore->new_memfd($name, $max, $initial);
my $sem = Data::Sync::Shared::Semaphore->new_from_fd($fd);

$max is the maximum permit count. $initial defaults to $max (fully available); set to 0 to start with no available permits.

Operations

my $ok  = $sem->acquire;              # block until available (infinite)
my $ok  = $sem->acquire($timeout);    # block with timeout (seconds)
my $ok  = $sem->try_acquire;          # non-blocking, false if unavailable
my $ok  = $sem->acquire_n($n);        # acquire N permits atomically
my $ok  = $sem->acquire_n($n, $timeout);
my $ok  = $sem->try_acquire_n($n);    # non-blocking N-permit acquire
$sem->release;                        # release one permit
$sem->release($n);                    # release N permits (clamped at max)
my $n   = $sem->drain;               # acquire all available, return count
my $val = $sem->value;                # current available count
my $max = $sem->max;                  # maximum permits

Guard

my $g = $sem->acquire_guard;          # acquire 1, release on scope exit
my $g = $sem->acquire_guard($n);      # acquire N
my $g = $sem->acquire_guard($n, $timeout);  # with timeout, undef on fail

Data::Sync::Shared::Barrier

Constructors

my $bar = Data::Sync::Shared::Barrier->new($path, $parties);
my $bar = Data::Sync::Shared::Barrier->new(undef, $parties);
my $bar = Data::Sync::Shared::Barrier->new_memfd($name, $parties);
my $bar = Data::Sync::Shared::Barrier->new_from_fd($fd);

$parties must be >= 2.

Operations

my $r = $bar->wait;           # block until all parties arrive
my $r = $bar->wait($timeout); # with timeout

Returns: 1 = leader (last to arrive), 0 = non-leader, -1 = timeout. A zero timeout (wait(0)) is a non-blocking probe: it returns -1 if the party would have to block, without breaking the barrier. On a positive timeout that actually elapses the barrier is permanently broken: wait returns -1 for the timing-out party and all other waiters, and every subsequent wait also returns -1 until reset is called. This matches POSIX pthread_barrier_t broken semantics.

my $gen = $bar->generation;    # how many times barrier has tripped
my $n   = $bar->arrived;       # currently arrived count
my $n   = $bar->parties;       # party count
$bar->is_broken;               # true if timeout broke the barrier
$bar->reset;                   # clear broken state + bump generation

Data::Sync::Shared::RWLock

Constructors

my $rw = Data::Sync::Shared::RWLock->new($path);
my $rw = Data::Sync::Shared::RWLock->new(undef);
my $rw = Data::Sync::Shared::RWLock->new_memfd($name);
my $rw = Data::Sync::Shared::RWLock->new_from_fd($fd);

Operations

$rw->rdlock;                   # block until read lock acquired
$rw->rdlock($timeout);         # with timeout (croaks on timeout)
$rw->wrlock;                   # block until write lock acquired
$rw->wrlock($timeout);         # with timeout (croaks on timeout)
my $ok = $rw->try_rdlock;      # non-blocking
my $ok = $rw->try_wrlock;      # non-blocking
my $ok = $rw->rdlock_timed($timeout);  # returns false on timeout
my $ok = $rw->wrlock_timed($timeout);  # returns false on timeout
$rw->rdunlock;
$rw->wrunlock;
$rw->downgrade;                # convert wrlock to rdlock atomically

Guards

my $g = $rw->rdlock_guard;             # rdunlock on scope exit
my $g = $rw->rdlock_guard($timeout);   # with timeout (croaks on fail)
my $g = $rw->wrlock_guard;
my $g = $rw->wrlock_guard($timeout);

Data::Sync::Shared::Condvar

Constructors

my $cv = Data::Sync::Shared::Condvar->new($path);
my $cv = Data::Sync::Shared::Condvar->new(undef);
my $cv = Data::Sync::Shared::Condvar->new_memfd($name);
my $cv = Data::Sync::Shared::Condvar->new_from_fd($fd);

Operations

$cv->lock;                     # acquire built-in mutex
$cv->unlock;                   # release built-in mutex
my $ok = $cv->try_lock;        # non-blocking

my $ok = $cv->wait;            # unlock, wait for signal, re-lock
my $ok = $cv->wait($timeout);  # with timeout
$cv->signal;                   # wake one waiter
$cv->broadcast;                # wake all waiters

my $ok = $cv->wait_while(\&pred);           # loop until pred returns false
my $ok = $cv->wait_while(\&pred, $timeout); # with timeout

wait must be called while holding the mutex. Returns 1 on signal/broadcast, 0 on timeout. The mutex is always re-acquired before wait returns.

wait_while calls wait in a loop until the predicate coderef returns false. Returns 1 if predicate became false, 0 on timeout.

Guard

my $g = $cv->lock_guard;       # unlock on scope exit

Data::Sync::Shared::Once

Constructors

my $once = Data::Sync::Shared::Once->new($path);
my $once = Data::Sync::Shared::Once->new(undef);
my $once = Data::Sync::Shared::Once->new_memfd($name);
my $once = Data::Sync::Shared::Once->new_from_fd($fd);

Operations

my $init = $once->enter;             # try + wait, infinite
my $init = $once->enter($timeout);   # with timeout
$once->done;                         # mark initialization complete
my $ok  = $once->is_done;            # check without blocking
$once->reset;                        # reset to uninitialized state

enter returns true for exactly one process (the initializer). All others block until done is called, then return false. With a $timeout, a waiter that times out before initialization completes also returns false: a false return means "not the initializer", not necessarily "initialization complete", so use is_done to tell a finished gate from a timed-out wait. If the initializer dies, stale PID detection elects a new one.

Common Methods

All primitives support:

my $p  = $obj->path;           # backing file path (undef if anon)
my $fd = $obj->memfd;          # memfd fd (-1 if file-backed/anon)
$obj->sync;                    # msync -- flush to disk
$obj->unlink;                  # remove backing file
Class->unlink($path);          # class method form
my $s  = $obj->stats;          # diagnostic hashref

Stats keys vary by type. All counters are approximate under concurrency.

Semaphore: value, max, waiters, mmap_size, acquires, releases, waits, timeouts, recoveries.

Barrier: parties, arrived, generation, waiters, mmap_size, waits, releases, timeouts.

RWLock: state ("unlocked", "read_locked", "write_locked"), readers, waiters, mmap_size, acquires, releases, recoveries.

Condvar: waiters, signals, mmap_size, acquires, releases, waits, timeouts, recoveries.

Once: state ("init", "running", "done"), is_done, waiters, mmap_size, acquires, releases, waits, timeouts, recoveries.

eventfd Integration

my $fd = $obj->eventfd;        # create eventfd, returns fd
$obj->eventfd_set($fd);        # use existing fd (e.g. from fork)
my $fd = $obj->fileno;         # current eventfd (-1 if none)
$obj->notify;                  # signal eventfd
my $n  = $obj->eventfd_consume;  # drain notification counter

Notification is opt-in. Use with EV or other event loops.

SEE ALSO

Data::Buffer::Shared - typed shared array

Data::HashMap::Shared - concurrent hash table

Data::Queue::Shared - FIFO queue

Data::PubSub::Shared - publish-subscribe ring

Data::ReqRep::Shared - request-reply

Data::Pool::Shared - fixed-size object pool

Data::Stack::Shared - LIFO stack

Data::Deque::Shared - double-ended queue

Data::Log::Shared - append-only log (WAL)

Data::Heap::Shared - priority queue

Data::Graph::Shared - directed weighted graph

Data::BitSet::Shared - shared bitset (lock-free per-bit ops)

Data::RingBuffer::Shared - fixed-size overwriting ring buffer

AUTHOR

vividsnow

LICENSE

This is free software; you can redistribute it and/or modify it under the same terms as Perl itself.