view release on metacpan or  search on metacpan

lib/Data/Queue/Shared.pm  view on Meta::CPAN

        $child->push(99);
        exit;
    }
    wait;
    print $q->pop;  # 99

=head1 DESCRIPTION

Data::Queue::Shared provides bounded MPMC (multi-producer, multi-consumer)
queues stored in file-backed shared memory (C<mmap(MAP_SHARED)>), enabling
efficient multiprocess data sharing on Linux.

B<Linux-only>. Requires 64-bit Perl.

=head2 Variants

=over

=item L<Data::Queue::Shared::Int> - int64 values, lock-free (16 bytes/slot)

Uses the Vyukov bounded MPMC algorithm. Push and pop are lock-free
(CAS-based). Optimal for integer job IDs, counters, indices.

=item L<Data::Queue::Shared::Int32> - int32 values, lock-free (8 bytes/slot)

=item L<Data::Queue::Shared::Int16> - int16 values, lock-free (8 bytes/slot)

Compact variants with 32-bit Vyukov sequence numbers. Half the memory
footprint per slot = double the cache density. Same lock-free algorithm.
Same API as Int. Values outside the type range are silently truncated
(standard C cast semantics).

=item L<Data::Queue::Shared::Str> - byte string values, mutex-protected

Uses a futex-based mutex with a circular arena for variable-length string
storage. Supports UTF-8 flag preservation. Optimal for messages,
serialized data, filenames.

Memory-efficient for mixed lengths (short + occasional long strings share
the arena), but throughput degrades under heavy multi-producer contention
because all pushes serialize on one mutex.

B<For fixed-length FIFO workloads that need maximum throughput:> use
L<Data::Deque::Shared::Str> with only C<push_back> / C<pop_front>. It's
fixed-slot-per-entry (memory use = capacity x max_len), uses a per-slot
publication state machine instead of a shared mutex, and measures
1.3x-4.7x faster than Queue::Str depending on contention (1 vs 8 writers,
32-byte payloads, single box). Use that if your messages share an upper
bound and you want lock-free-style scaling.

=back

=head2 Features

=over

=item * File-backed mmap for cross-process sharing

=item * Lock-free MPMC for integer queues (Vyukov algorithm)

=item * Futex-based blocking wait with timeout (no busy-spin)

=item * PID-based stale lock recovery (dead process detection)

=item * Batch push/pop operations

=item * Circular arena for zero-fragmentation string storage

=item * Optional keyword API via XS::Parse::Keyword (zero method-dispatch overhead)

=back

=head2 Constructor

    # Int queue
    my $q = Data::Queue::Shared::Int->new($path, $capacity);

    # Str queue
    my $q = Data::Queue::Shared::Str->new($path, $capacity);
    my $q = Data::Queue::Shared::Str->new($path, $capacity, $arena_bytes);

Creates or opens a shared queue backed by file C<$path>.
C<$capacity> is rounded up to the next power of 2.
When opening an existing file, parameters are read from the stored header.
Multiple processes can open the same file simultaneously.

Pass C<undef> for C<$path> to create an anonymous queue using
C<MAP_SHARED|MAP_ANONYMOUS>. Anonymous queues are shared with child
processes via C<fork()> but cannot be opened by unrelated processes.

=head2 memfd Constructor

    my $q = Data::Queue::Shared::Int->new_memfd($name, $capacity);
    my $q = Data::Queue::Shared::Str->new_memfd($name, $capacity);
    my $q = Data::Queue::Shared::Str->new_memfd($name, $cap, $arena);

Creates a queue backed by C<memfd_create(2)>. No filesystem path -- the
backing memory is identified by a file descriptor. Use C<memfd()> to
retrieve the fd and pass it to other processes via C<SCM_RIGHTS>
(Unix domain socket fd passing) or C<fork()> inheritance.

    my $q2 = Data::Queue::Shared::Int->new_from_fd($fd);
    my $q2 = Data::Queue::Shared::Str->new_from_fd($fd);

Opens a queue from a received memfd. The fd is dup'd internally.

    my $fd = $q->memfd;    # backing fd (-1 if file-backed/anonymous)

For Str queues, C<$arena_bytes> sets the string storage arena size
(default: C<$capacity * 256>, minimum 4096, maximum 4GB). Strings are
stored in a circular arena; total stored string bytes cannot exceed the
arena capacity. Individual strings are limited to ~2GB.

=head2 API

=head3 Core operations

    my $ok  = $q->push($value);             # non-blocking, false if full
    my $val = $q->pop;                       # non-blocking, undef if empty
    my $ok  = $q->push_wait($value);         # blocking, infinite wait
    my $ok  = $q->push_wait($value, $secs);  # blocking with timeout