File Coverage

blib/lib/IPC/Shareable.pm

Criterion	Covered	Total	%
statement	1009	1061	95.1
branch	565	718	78.6
condition	167	250	66.8
subroutine	89	90	98.8
pod	22	22	100.0
total	1852	2141	86.5

package IPC::Shareable;

use warnings;
use strict;

require 5.010;

use Carp qw(croak confess carp);
use Config;
use Errno qw(EINVAL ENOMEM ENOSPC);
use Digest::MD5 qw(md5_hex);
use IPC::Semaphore;
use IPC::Shareable::SharedMem;
use IPC::SysV qw(
    IPC_PRIVATE
    IPC_CREAT
    IPC_EXCL
    IPC_NOWAIT
    IPC_RMID
    IPC_STAT
    SEM_UNDO
);
use JSON qw(-convert_blessed_universally);
use Scalar::Util;
use String::CRC32;
use Storable 0.6 qw(freeze thaw);

our $VERSION = '1.19';

# eval() returns 1 on success; // 0 coerces undef (failure) to 0 so callers
# can boolean-test cleanly without checking definedness.

our $_have_xs = ! $ENV{IPC_SHAREABLE_NO_XS} && eval {
    require XSLoader;
    XSLoader::load('IPC::Shareable', $VERSION);
    1;
} // 0;

use constant {
    # Locking

    LOCK_SH               => 1,
    LOCK_EX               => 2,
    LOCK_NB               => 4,
    LOCK_UN               => 8,

    # SHM parameters

    SHM_BUFSIZ            => 65536,
    SHMMAX_BYTES          => 1073741824, # ~1 GB
    SHM_EXISTS            => 1,

    # Semaphore slots (4 slots always; 5th slot added when 'testing' is set)

    SEM_MARKER            => 0,
    SEM_READERS           => 1,
    SEM_WRITERS           => 2,
    SEM_PROTECTED         => 3,
    SEM_TESTING           => 4,

    # Perl sends in a double as opposed to an integer to shmat(), and on some
    # systems, this causes the IPC system to round down to the maximum integer
    # size of 0x80000000. We correct that when generating keys with CRC32.

    MAX_KEY_INT_SIZE      => 0x80000000,

    # Number of times we'll check for existing segs

    EXCLUSIVE_CHECK_LIMIT => 10,

    # Struct types

    TYPE_HASH             => 0,
    TYPE_ARRAY            => 1,
    TYPE_SCALAR           => 2,
};

require Exporter;
our @ISA = 'Exporter';
our @EXPORT_OK = qw(
    LOCK_EX
    LOCK_SH
    LOCK_NB
    LOCK_UN
    SEM_MARKER
    SEM_READERS
    SEM_WRITERS
    SEM_PROTECTED
    SEM_TESTING
);
our %EXPORT_TAGS = (
    all         => [
        qw( LOCK_EX LOCK_SH LOCK_NB LOCK_UN ),
        qw( SEM_MARKER SEM_READERS SEM_WRITERS SEM_PROTECTED SEM_TESTING ),
    ],
    lock        => [qw( LOCK_EX LOCK_SH LOCK_NB LOCK_UN )],
    flock       => [qw( LOCK_EX LOCK_SH LOCK_NB LOCK_UN )],
    semaphores  => [qw( SEM_MARKER SEM_READERS SEM_WRITERS SEM_PROTECTED SEM_TESTING )],
);

# Locking scheme copied from IPC::ShareLite (with minor modifications)

my %semop_args = (
    (LOCK_EX),
    [
        SEM_READERS, 0, 0,                        # Wait for readers to finish
        SEM_WRITERS, 0, 0,                        # Wait for writers to finish
        SEM_WRITERS, 1, SEM_UNDO,                 # Assert write lock
    ],
    (LOCK_EX|LOCK_NB),
    [
        SEM_READERS, 0, IPC_NOWAIT,               # Wait for readers to finish
        SEM_WRITERS, 0, IPC_NOWAIT,               # Wait for writers to finish
        SEM_WRITERS, 1, (SEM_UNDO | IPC_NOWAIT),  # Assert write lock
    ],
    (LOCK_EX|LOCK_UN),
    [
        SEM_WRITERS, -1, (SEM_UNDO | IPC_NOWAIT),
    ],
    (LOCK_SH),
    [
        SEM_WRITERS, 0, 0,                        # Wait for writers to finish
        SEM_READERS, 1, SEM_UNDO,                 # Assert shared read lock
    ],
    (LOCK_SH|LOCK_NB),
    [
        SEM_WRITERS, 0, IPC_NOWAIT,               # Wait for writers to finish
        SEM_READERS, 1, (SEM_UNDO | IPC_NOWAIT),  # Assert shared read lock
    ],
    (LOCK_SH|LOCK_UN),
    [
        SEM_READERS, -1, (SEM_UNDO | IPC_NOWAIT), # Remove shared read lock
    ],
);

my %default_options = (
    key                         => IPC_PRIVATE,
    create                      => 0,
    exclusive                   => 0,
    destroy                     => 0,
    mode                        => 0666,
    size                        => SHM_BUFSIZ,
    protected                   => 0,
    testing                     => 0,
    limit                       => 1,
    graceful                    => 0,
    warn                        => 0,
    serializer                  => 'json',
    enforced_write_locking      => 1,
    enforced_read_locking       => 1,
    violated_write_lock_warn    => 1,
    violated_read_lock_warn     => 1,
);

# Class-level variables

my %global_register;
my %process_register;
my %used_ids;
my $_testing_dist = '';

# Set once we have warned that a semaphore set vanished mid-unlock (a peer
# removed it). Keeps the warning to one line per process rather than per call.

my $_unlock_einval_warned = 0;

# "Magic" methods

sub TIESCALAR {
    return _tie('SCALAR', @_);
}
sub TIEARRAY {
    return _tie('ARRAY', @_);
}
sub TIEHASH {
    return _tie('HASH', @_);
}
sub STORE {
    my $knot = shift;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg) unless ($knot->{_lock});

    if ($knot->{_type_int} == TYPE_HASH) {
        my ($key, $val) = @_;
        _remove_child($knot->{_data}{$key});
        _magic_tie($knot, $val) if ref($val) && $knot->_need_tie($val);
        $knot->{_data}{$key} = $val;
    }
    elsif ($knot->{_type_int} == TYPE_ARRAY) {
        my ($i, $val) = @_;
        _remove_child($knot->{_data}[$i]);
        _magic_tie($knot, $val) if ref($val) && $knot->_need_tie($val);
        $knot->{_data}[$i] = $val;
    }
    elsif ($knot->{_type_int} == TYPE_SCALAR) {
        my ($val) = @_;

        if ($knot->{_data} && ref($knot->{_data})) {
            _remove_child(${$knot->{_data}});
        }
        _magic_tie($knot, $val) if ref($val) && $knot->_need_tie($val);
        $knot->{_data} = \$val;
    }

    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }

    return 1;
}
sub FETCH {
    my $knot = shift;

    my $data;

    if ($knot->{_lock}) {
        $data = $knot->{_data};
    }
    else {
        _read_check($knot);
        $data = $knot->_decode($knot->seg);
        $knot->{_data} = $data;
    }

    my $val;

    if ($knot->{_type_int} == TYPE_HASH) {
        my $key = shift;
        $val = $data->{$key};
    }
    elsif ($knot->{_type_int} == TYPE_ARRAY) {
        my $i = shift;
        $val = $data->[$i];
    }
    elsif ($knot->{_type_int} == TYPE_SCALAR) {
        if (defined $data) {
            $val = $$data;
        }
        else {
            return;
        }
    }

    if (ref($val) && (my $inner = _is_child($val))) {
        # Register the inner knot so clean_up_all() can find it even when it
        # was created in a forked child process

        if (! exists $global_register{$inner->seg->id}) {
            $global_register{$inner->seg->id} = $inner;
        }

        unless ($inner->{_lock}) {
            my $s = $inner->seg;
            $inner->{_data} = $knot->_decode($s);
        }
    }
    return $val;

}
sub CLEAR {
    my $knot = shift;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg) unless $knot->{_lock};

    if ($knot->{_type_int} == TYPE_HASH) {
        for my $val (values %{ $knot->{_data} }) {
            _remove_child($val);
        }
        $knot->{_data} = { };
    }
    elsif ($knot->{_type_int} == TYPE_ARRAY) {
        for my $val (@{ $knot->{_data} }) {
            _remove_child($val);
        }
        $knot->{_data} = [ ];
    }

    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
}
sub DELETE {
    my $knot = shift;
    my $key  = shift;

    croak "Cannot delete from a non-hash tied variable"
        unless $knot->{_type_int} == TYPE_HASH;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg) unless $knot->{_lock};
    my $val = delete $knot->{_data}->{$key};

    _remove_child($val);

    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }

    return $val;
}
sub EXISTS {
    my $knot = shift;
    my $key  = shift;

    $knot->{_data} = $knot->_decode($knot->seg) unless $knot->{_lock};
    return exists $knot->{_data}->{$key};
}
sub FIRSTKEY {
    my $knot = shift;
    $knot->{_data} = $knot->_decode($knot->seg) unless $knot->{_lock};
    $knot->{_hkey_list} = [ keys %{$knot->{_data}} ];
    return $knot->NEXTKEY;
}
sub NEXTKEY {
    my ($knot, $last_key_accessed) = @_;

    # We don't use ordered hashes, so we don't need to use
    # the last key accessed parameter

    # Caveat emptor if hash was changed by another process

    return shift @{$knot->{_hkey_list}};
}
sub EXTEND {
    #XXX Noop
}
sub PUSH {
    my $knot = shift;

    croak "Cannot push to a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg, $knot->{_data}) unless $knot->{_lock};

    push @{$knot->{_data}}, @_;
    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
}
sub POP {
    my $knot = shift;

    croak "Cannot pop from a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg, $knot->{_data}) unless $knot->{_lock};

    my $val = pop @{$knot->{_data}};
    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
    return $val;
}
sub SHIFT {
    my $knot = shift;

    croak "Cannot shift from a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg, $knot->{_data}) unless $knot->{_lock};
    my $val = shift @{$knot->{_data}};
    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
    return $val;
}
sub UNSHIFT {
    my $knot = shift;

    croak "Cannot unshift a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg, $knot->{_data}) unless $knot->{_lock};
    my $val = unshift @{$knot->{_data}}, @_;
    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
    return $val;
}
sub SPLICE {
    my($knot, $off, $n, @av) = @_;

    croak "Cannot splice a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg, $knot->{_data}) unless $knot->{_lock};
    my @val = splice @{$knot->{_data}}, $off, $n, @av;
    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
    return @val;
}
sub FETCHSIZE {
    my $knot = shift;

    croak "Cannot fetchsize on a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    $knot->{_data} = $knot->_decode($knot->seg) unless $knot->{_lock};
    return scalar(@{$knot->{_data}});
}
sub STORESIZE {
    my $knot = shift;
    my $n    = shift;

    croak "Cannot storesize on a non-array tied variable"
        unless $knot->{_type_int} == TYPE_ARRAY;

    return if ! _write_permitted($knot);

    $knot->{_data} = $knot->_decode($knot->seg) unless $knot->{_lock};
    $#{$knot->{_data}} = $n - 1;

    if ($knot->{_lock} & LOCK_EX) {
        $knot->{_was_changed} = 1;
    }
    else {
        _write_to_seg($knot);
    }
    return $n;
}

# Public methods

*shlock = \&lock;
*shunlock = \&unlock;

# End user methods

sub new {
    my ($class, %opts) = @_;

    my $type = $opts{var} || 'HASH';

    if ($type eq 'HASH') {
        tie my %h, 'IPC::Shareable', \%opts;
        return \%h;
    }
    if ($type eq 'ARRAY') {
        tie my @a, 'IPC::Shareable', \%opts;
        return \@a;
    }
    if ($type eq 'SCALAR') {
        tie my $s, 'IPC::Shareable', \%opts;
        return \$s;
    }
}
sub lock {
    my $knot = shift;

    my ($flags, $code);

    if (scalar @_ == 2) {
        ($flags, $code) = @_;
    }

    if (defined $_[0]) {
        if (ref $_[0] eq 'CODE') {
            $code = shift;
        }
        else {
            $flags = shift;
        }
    }

    if (defined $code && ref $code ne 'CODE') {
        croak "\$code param to lock() must be a code reference";
    }

    $flags = LOCK_EX if ! defined $flags;

    # unlock() was called

    return $knot->unlock if ($flags & LOCK_UN);

    # Caller already has the same lock type

    if ($knot->{_lock} & $flags) {
        if ($code && $flags == LOCK_EX) {
            _execute_lock_coderef($knot, $code);
        }
        return 1;
    }

    # If they have a different lock than they want, release it first

    $knot->unlock if ($knot->{_lock});

    my $sem = $knot->sem;
    my $lock_success = $sem->op(@{ $semop_args{$flags} });

    if ($lock_success) {
        $knot->{_lock} = $flags;
        $knot->{_data} = $knot->_decode($knot->seg);

        my $locked_ref = _lock_children($knot, $flags);

        if (! $locked_ref) {
            my $rflags = $knot->{_lock} | LOCK_UN;
            $rflags ^= LOCK_NB if $rflags & LOCK_NB;
            $knot->sem->op(@{ $semop_args{$rflags} });
            $knot->{_lock} = 0;
            $lock_success  = 0;
        }
        else {
            $knot->{_locked_children} = $locked_ref;
        }
    }

    if ($flags == LOCK_EX && $lock_success && $code) {
        _execute_lock_coderef($knot, $code);
        return 1;
    }
    return $lock_success;
}
sub unlock {
    my $knot = shift;

    return 1 unless $knot->{_lock};

    if ($knot->{_was_changed}) {
        _write_to_seg($knot);
        $knot->{_was_changed} = 0;
    }

    # Unlock children/nested segs in reverse order

    for my $child (reverse @{ $knot->{_locked_children} // [] }) {
        if ($child->{_was_changed}) {
            _write_to_seg($child);
            $child->{_was_changed} = 0;
        }

        my $child_flags = $child->{_lock} | LOCK_UN;

        $child_flags ^= LOCK_NB if $child_flags & LOCK_NB;
        $child->sem->op(@{ $semop_args{$child_flags} });
        $child->{_lock} = 0;
    }

    $knot->{_locked_children} = [];

    # Release semaphore locks

    my $sem = $knot->sem;
    my $flags = $knot->{_lock} | LOCK_UN;

    $flags ^= LOCK_NB if ($flags & LOCK_NB);

    if (! $sem->op(@{ $semop_args{$flags} })) {
        if ($!{EINVAL}) {
            # The semaphore set was removed by another process (eg. a peer
            # holding the same segment with destroy=>1 exited). The lock we
            # held went away with it, so there is nothing left to release.
            # Warn once and carry on rather than aborting the caller; every
            # other errno is a real failure and stays fatal. This mirrors the
            # read path's tolerance of an unreachable set (see _write_permitted
            # and _check_read_lock).

            carp "Semaphore set gone during unlock (removed by another "
               . "process); treating the lock as already released"
                if ! $_unlock_einval_warned;

            $_unlock_einval_warned = 1;
        }
        else {
            croak "Could not release semaphore lock: $!\n";
        }
    }

    $knot->{_lock} = 0;

    1;
}
sub singleton {

    # If called with IPC::Shareable::singleton() as opposed to
    # IPC::Shareable->singleton(), the class isn't sent in. Check
    # for this and fix it if necessary

    if (! defined $_[0] || $_[0] ne __PACKAGE__) {
        unshift @_, __PACKAGE__;
    }

    my ($class, $glue, $warn) = @_;

    if (! defined $glue) {
        croak "singleton() requires a GLUE parameter";
    }

    $warn = 0 if ! defined $warn;

    tie my $lock, 'IPC::Shareable', {
        key         => $glue,
        create      => 1,
        exclusive   => 1,
        graceful    => 1,
        destroy     => 1,
        warn        => $warn
    };

    return $$;
}

# Helper, maintenance and developer methods

sub attributes {
    my ($knot, $attr) = @_;

    if (defined $attr) {
        return $knot->{attributes}{$attr};
    }
    else {
        return $knot->{attributes};
    }
}
sub global_register {
    return \%global_register;
}
sub process_register {
    return \%process_register;
}
sub uuid {
    my ($knot) = @_;

    if (! defined $knot->{_uuid}) {
        $knot->{_uuid} = md5_hex(rand());
    }

    return $knot->{_uuid};
}

sub seg {
    my ($knot) = @_;
    return $knot->{_shm} if defined $knot->{_shm};
}
sub sem {
    my ($knot) = @_;
    return $knot->{_sem} if defined $knot->{_sem};
}

sub shm_segments {
    shift if ref($_[0]) || (defined $_[0] && ! ref($_[0]) && UNIVERSAL::isa($_[0], __PACKAGE__));

    my ($filter_key) = @_;

    my $filter_int = _key_str_to_int($filter_key) if defined $filter_key;

    my %segments;

    open my $ipcs_fh, '-|', 'ipcs', '-m' or die "ipcs -m: $!";

    while (my $line = <$ipcs_fh>) {
        my ($id, $raw_key);

        if ($line =~ /^\s*m\s+(\d+)\s+(\S+)/) {
            # BSD/macOS format: m   ...
            ($id, $raw_key) = ($1, $2);
        }
        elsif ($line =~ /^\s*(\d+)\s+(0x[0-9a-fA-F]+)\s+/) {
            # DragonFly BSD format:   ... (no 'm' type column)
            ($id, $raw_key) = ($1, $2);
        }
        elsif ($line =~ /^\s*(\S+)\s+(\d+)\s+\S+/) {
            # Linux format:   ...
            ($raw_key, $id) = ($1, $2);
        }
        else {
            next;
        }

        my $key_int = $raw_key =~ /^0x[0-9a-fA-F]+$/
            ? hex($raw_key)
            : $raw_key =~ /^\d+$/
            ? int($raw_key)
            : next;

        my $hex_key = sprintf('0x%08x', $key_int);

        next if $key_int == 0;  # IPC_PRIVATE segments can't be found by key

        # Get segment size via IPC_STAT

        my $stat_buf = '';
        shmctl($id, IPC_STAT, $stat_buf) or next;

        my ($segsz) = $^O eq 'linux'
            ? ( $Config{longsize} == 8
            ? unpack('x[48] Q', $stat_buf)   # 64-bit Linux
            : unpack('x[36] L', $stat_buf) ) # 32-bit Linux
            : $^O eq 'freebsd' && $Config{longsize} == 8
            ? unpack('x[32] Q', $stat_buf)   # 64-bit FreeBSD (key_t=long=8, ipc_perm=32)
            : $^O eq 'solaris'
            ? ( $Config{longsize} == 8
            ? unpack('x[32] Q', $stat_buf)   # 64-bit Solaris (ipc_perm=28 + pad 4)
            : unpack('x[44] L', $stat_buf) ) # 32-bit Solaris (ipc_perm=44)
            : $^O eq 'openbsd' && $Config{longsize} == 8
            ? unpack('x[32] L', $stat_buf)   # 64-bit OpenBSD: segsz is int (4 bytes)
            : $^O eq 'dragonfly' && $Config{longsize} == 8
            ? unpack('x[32] Q', $stat_buf)   # 64-bit DragonFly (ipc_perm=28 + pad 4; segsz=size_t=8)
            : unpack('x[24] Q', $stat_buf);  # macOS

        next unless $segsz;

        # Probe the 14-byte tag first so we don't pull entire foreign
        # segments (which may be gigabytes) into Perl just to discard them.

        my $head = '';
        shmread($id, $head, 0, 14) or next;
        next unless $head eq 'IPC::Shareable';

        my $data = '';
        shmread($id, $data, 0, $segsz) or next;

        # Strip trailing null bytes
        $data =~ s/\x00+$//;

        my $json_part  = substr($data, 14);
        my @child_keys = ($json_part =~ /"child_key_hex":"([^"]+)"/g);

        $segments{$hex_key} = {
            child_keys    => \@child_keys,
            content       => $data,
            id            => $id,
            local_process => (exists $process_register{$id} ? 1 : 0),
            known         => (exists $global_register{$id}  ? 1 : 0),
        };
    }
    close $ipcs_fh;

    if (defined $filter_int) {
        # Walk the segment tree starting from the root whose key matches
        # $filter_int, collecting it and all its descendants.  Use integer
        # comparison so that hex formatting differences (zero-padding, case)
        # between ipcs(1) output and child_key_hex values don't matter.

        my %int_to_hex = map { hex($_) => $_ } keys %segments;

        my (%related, @queue);

        push @queue, $filter_int;

        while (my $k_int = shift @queue) {
            my $k_hex = $int_to_hex{$k_int} // next;
            next if $related{$k_hex}++;
            push @queue, map { hex($_) } @{ $segments{$k_hex}{child_keys} };
        }

        %segments = map { $_ => $segments{$_} } keys %related;
    }

    return \%segments;
}
sub unknown_segments {
    shift if ref $_[0]; # Allow for object or class method call

    my $segs = shm_segments();

    return grep { ! $segs->{$_}{known} } keys %$segs;
}
sub seg_count {
    my $count = 0;

    open my $ipcs_fh, '-|', 'ipcs', '-m' or die "ipcs -m: $!";

    while (my $line = <$ipcs_fh>) {
        if ($line =~ /^\s*m\s+\d+\s+\S+/) {
            # BSD/macOS format: m   ...
            $count++;
        }
        elsif ($line =~ /^\s*\d+\s+0x[0-9a-fA-F]+\s+/) {
            # DragonFly BSD:   ... (no type-letter column)
            $count++;
        }
        elsif ($line =~ /^\s*(?:0x[0-9a-fA-F]+|\d+)\s+\d+\s+\S+/) {
            # Linux format:   ...
            $count++;
        }
    }

    close $ipcs_fh;

    return $count;
}
sub sem_count {
    my $count = 0;

    open my $ipcs_fh, '-|', 'ipcs', '-s' or die "ipcs -s: $!";

    while (my $line = <$ipcs_fh>) {
        if ($line =~ /^\s*s\s+\d+\s+\S+/) {
            # BSD/macOS format: s   ...
            $count++;
        }
        elsif ($line =~ /^\s*\d+\s+0x[0-9a-fA-F]+\s+/) {
            # DragonFly BSD:   ... (no type-letter column)
            $count++;
        }
        elsif ($line =~ /^\s*(?:0x[0-9a-fA-F]+|\d+)\s+\d+\s+\S+/) {
            # Linux format:   ...
            $count++;
        }
    }

    close $ipcs_fh;

    return $count;
}
sub seg_map {
    croak "seg_map() must be called as an object method" unless ref $_[0];
    my $knot_filter = shift;

    my $segs = shm_segments();

    # Build hex_key -> OS segment ID from shm_segments() data

    my %id_by_hex;
    $id_by_hex{ $_ } = $segs->{$_}{id} for keys %$segs;

    # Build hex_key -> knot from global_register (keyed by seg_id)

    my %knot_by_hex;
    for my $id (keys %global_register) {
        my $knot = $global_register{$id};
        my $hex  = $knot->{_key_hex};

        $knot_by_hex{$hex} = $knot if defined $hex;
    }

    # Supplement child_keys from global_register for Storable segments.
    # shm_segments() only extracts child_key_hex from JSON segment content;
    # for Storable we walk each knot's _data looking for tied child references

    my %extra_child_keys;   # hex_key -> [ child_hex, ... ]

    for my $hex (keys %knot_by_hex) {
        my $knot  = $knot_by_hex{$hex};
        my $data  = $knot->{_data};
        my $rtype = Scalar::Util::reftype($data) // '';

        my @vals = $rtype eq 'HASH'  ? values %$data
                 : $rtype eq 'ARRAY' ? @$data
                 : ();

        for my $v (@vals) {
            next unless ref($v);

            my $vtype = Scalar::Util::reftype($v) // '';
            my $child_knot;

            if ($vtype eq 'HASH')   {
                $child_knot = tied(%$v)
            }
            elsif ($vtype eq 'ARRAY')  {
                $child_knot = tied(@$v)
            }
            elsif ($vtype eq 'SCALAR') {
                $child_knot = tied($$v)
            }

            next unless $child_knot && $child_knot->{_key_hex};

            push @{ $extra_child_keys{$hex} }, $child_knot->{_key_hex};
        }
    }

    # If called as an object method, restrict output to just that knot's tree
    # by BFS through both child_keys (JSON) and extra_child_keys (Storable).

    if ($knot_filter && $knot_filter->{_key_hex}) {
        my $root_hex = $knot_filter->{_key_hex};
        my (%in_tree, @queue);

        push @queue, $root_hex;

        while (my $h = shift @queue) {
            next if $in_tree{$h}++;

            push @queue, @{ $segs->{$h}{child_keys}   // [] };
            push @queue, @{ $extra_child_keys{$h}     // [] };
        }

        %$segs = map { $_ => $segs->{$_} } grep { $in_tree{$_} } keys %$segs;
    }

    # Identify root segments (not a child of any other segment)

    my %is_child;

    for my $hex (keys %$segs) {
        $is_child{$_}++ for @{ $segs->{$hex}{child_keys} };
    }

    for my $hex (keys %extra_child_keys) {
        next unless exists $segs->{$hex};
        $is_child{$_}++ for @{ $extra_child_keys{$hex} };
    }

    my @roots = sort grep { ! $is_child{$_} } keys %$segs;

    my @lines;
    push @lines, 'IPC::Shareable Segment Map';
    push @lines, '=' x 26;

    if (! @roots) {
        push @lines, '';
        push @lines, '  (no IPC::Shareable segments found)';
        return join("\n", @lines) . "\n";
    }

    my $render;

    $render = sub {
        my ($hex, $depth) = @_;
        my $indent = '  ' x $depth;
        my $seg    = $segs->{$hex} // {};

        my @tags;
        push @tags, $seg->{known} ? 'known' : 'unknown';
        push @tags, 'owner' if $seg->{local_process};
        my $tag_str = '[' . join(', ', @tags) . ']';

        my $seg_id = $id_by_hex{$hex} // '?';

        # Read semaphore slot values and ID; for segments not in
        # global_register attach with nsems=0 (avoids EINVAL on existing sets)

        my ($sem_str, $content_str);

        my $sem = $knot_by_hex{$hex}
            ? $knot_by_hex{$hex}->sem
            : IPC::Semaphore->new(hex($hex), 0, 0);

        if (defined $sem) {
            my $sem_id    = $sem->id                    // '?';
            my $marker    = $sem->getval(SEM_MARKER)    // '?';
            my $readers   = $sem->getval(SEM_READERS)   // '?';
            my $writers   = $sem->getval(SEM_WRITERS)   // '?';
            my $protected = $sem->getval(SEM_PROTECTED) // '?';

            # Continuation indent: one tab (8 spaces) from the left margin
            my $cont = ' ' x 8;

            $sem_str = join("\n",
                "sem_id: $sem_id",
                "${cont}1: SEM_MARKER=$marker",
                "${cont}2: READERS=$readers",
                "${cont}3: WRITERS=$writers",
                "${cont}4: PROTECTED=$protected",
            );
        }
        else {
            $sem_str = '(not accessible)';
        }

        $content_str = $knot_by_hex{$hex}
            ? _shm_data_summary($knot_by_hex{$hex})
            : '(not accessible - segment not tied in this process)';

        # Merge child keys from shm_segments() and from global_register walk

        my %seen_child;

        my @child_keys = grep { ! $seen_child{$_}++ } (
            @{ $seg->{child_keys} // [] },
            @{ $extra_child_keys{$hex} // [] },
        );

        my $children = @child_keys ? join(', ', @child_keys) : '(none)';

        push @lines, '';
        push @lines, "${indent}${tag_str}  key: ${hex}  seg_id: ${seg_id}";
        push @lines, "${indent}  Semaphores: ${sem_str}";
        push @lines, "${indent}  Children:   ${children}";
        push @lines, "${indent}  Content:    ${content_str}";

        $render->($_, $depth + 1) for @child_keys;
    };

    $render->($_, 0) for @roots;

    push @lines, '';

    return join("\n", @lines) . "\n";
}
sub sysv_info {
    shift; # Discard invocant (object ref or class name)

    my %opts       = @_;

    my $proc_dir   = delete $opts{_proc_dir}   // '/proc/sys/kernel';
    my $sysctl_out = delete $opts{_sysctl_out};

    my %info;

    if ($^O eq 'darwin') {
        my $out = defined $sysctl_out ? $sysctl_out : do {
            open my $fh, '-|', 'sysctl', 'kern.sysv' or die "sysctl: $!";
            local $/;
            my $s = <$fh>;
            close $fh;
            $s;
        };
        for my $line (split /\n/, $out) {
            if ($line =~ /^kern\.sysv\.(\w+):\s*(\S+)/) {
                $info{$1} = $2;
            }
        }
    }
    elsif ($^O eq 'freebsd' || $^O eq 'midnightbsd' || $^O eq 'netbsd') {
        # MidnightBSD (FreeBSD-derived) and NetBSD share FreeBSD's kern.ipc
        # sysctl namespace. NetBSD's semmni also defaults to 10, so exposing
        # the limit there lets the test suite's free-set guard activate
        # instead of dying ENOSPC mid-run.

        my $out = defined $sysctl_out ? $sysctl_out : do {
            open my $fh, '-|', 'sysctl', 'kern.ipc' or die "sysctl: $!";
            local $/;
            my $s = <$fh>;
            close $fh;
            $s;
        };
        for my $line (split /\n/, $out) {
            if ($line =~ /^kern\.ipc\.((?:shm|sem)\w+)\s*[:=]\s*(\S+)/) {
                $info{$1} = $2;
            }
        }
    }
    elsif ($^O eq 'openbsd') {
        my $out = defined $sysctl_out ? $sysctl_out : do {
            open my $fh, '-|', 'sysctl', 'kern.seminfo', 'kern.shminfo'
                or die "sysctl: $!";
            local $/;
            my $s = <$fh>;
            close $fh;
            $s;
        };
        for my $line (split /\n/, $out) {
            if ($line =~ /^kern\.(?:sem|shm)info\.(\w+)\s*=\s*(\S+)/) {
                $info{$1} = $2;
            }
        }
    }
    elsif ($^O eq 'linux') {
        for my $key (qw(shmmax shmmin shmmni shmall)) {
            my $file = "$proc_dir/$key";
            if (open my $fh, '<', $file) {
                chomp(my $val = <$fh>);
                $info{$key} = $val;
            }
        }
        # /proc/sys/kernel/sem is a single line of 4 ints:
        #   semmsl semmns semopm semmni
        if (open my $fh, '<', "$proc_dir/sem") {
            chomp(my $line = <$fh>);
            close $fh;
            my @vals = split /\s+/, $line;
            if (@vals >= 4) {
                @info{qw(semmsl semmns semopm semmni)} = @vals[0..3];
            }
        }
    }

    return %info ? \%info : undef;
}

# Cleanup

sub clean_up {
    my $class = shift;

    for my $id (keys %process_register) {
        my $s = $process_register{$id};
        next unless $s->attributes('owner') == $$;
        next if $s->attributes('protected');
        remove($s);
    }
}
sub clean_up_all {
    my $class = shift;

    my $global_register = __PACKAGE__->global_register;

    for my $id (keys %$global_register) {
        my $s = $global_register->{$id};
        next if $s->attributes('protected');
        remove($s);
    }
}
sub clean_up_protected {
    my ($knot, $protect_key);

    if (scalar @_ == 2) {
        ($knot, $protect_key) = @_;
    }
    if (scalar @_ == 1) {
        ($protect_key) = @_;
    }

    if (! defined $protect_key) {
        croak "clean_up_protected() requires a \$protect_key param";
    }

    if ($protect_key !~ /^\d+$/) {
        croak
            "clean_up_protected() \$protect_key must be an integer. You sent $protect_key";
    }

    my $global_register = __PACKAGE__->global_register;

    for my $id (keys %$global_register) {
        my $s = $global_register->{$id};
        my $stored_key = $s->attributes('protected');

        if ($stored_key && $stored_key == $protect_key) {
            remove($s);
        }
    }
}
sub remove {
    my ($knot, $key) = @_;

    # If a key is passed, remove that specific segment by key rather than
    # via an existing tied object

    if (defined $key) {
        $key = $knot->_shm_key($key);
        my $id = shmget($key, 0, 0);

        if (! defined $id) {
            warn "remove(): shmget failed for key $key: $!";
            return;
        }

        if (! shmctl($id, IPC_RMID, 0)) {
            warn "Couldn't remove shm segment $id: $!";
        }
        else {
            delete $process_register{$id};
            delete $global_register{$id};
        }

        # Remove the associated semaphore set (same key, attach-only with nsems=0)

        my $sem = IPC::Semaphore->new($key, 0, 0);
        if (defined $sem) {
            $sem->remove or warn "Couldn't remove semaphore set for key $key: $!";
        }

        return;
    }

    # Standard object based removal

    my $seg = $knot->seg;
    my $id = $seg->id;

    my $seg_removed = 0;

    if (! $seg->remove) {
        warn "Couldn't remove shm segment $id: $!";
    }
    else {
        $seg_removed = 1;
    }

    # Semaphore cleanup

    my $sem = $knot->sem;

    my $sem_removed = 0;
    my $sem_remove_status = $sem->remove;

    if ($sem_remove_status != 1 && $sem_remove_status ne '0 but true') {
        warn "Couldn't remove semaphore set $id: $!";
    }
    else {
        $sem_removed = 1;
    }

    # If the segment or semaphore couldn't be cleaned up, we need to
    # keep state

    if ($seg_removed && $sem_removed) {
        delete $process_register{$id};
        delete $global_register{$id};
    }
}

# Unit testing

sub testing_set {
    my ($class, $dist_name) = @_;

    croak "testing_set() requires a distribution name string"
        unless defined $dist_name && length $dist_name;

    $_testing_dist = $dist_name;
}
sub clean_up_testing {
    shift if @_ > 1 && ! ref $_[0] && defined $_[0] && UNIVERSAL::isa($_[0], __PACKAGE__);

    my ($dist_name) = @_;

    croak "clean_up_testing() requires a distribution name string"
        unless defined $dist_name && length $dist_name;

    my $target  = _testing_semaphore_key_hash($dist_name);
    my $removed = 0;

    # Scan ipcs -m for segment IDs and keys directly. We cannot use
    # shm_segments() here because it filters by the 'IPC::Shareable' 14-byte
    # tag, which is only written during STORE operations — empty tied segments
    # have no tag and would be invisible. The authoritative identifier for a
    # testing-tagged segment is the SEM_TESTING value on its semaphore set,
    # not the segment content.

    open my $ipcs_fh, '-|', 'ipcs', '-m' or die "ipcs -m: $!";

    while (my $line = <$ipcs_fh>) {
        my ($id, $raw_key);

        if ($line =~ /^\s*m\s+(\d+)\s+(\S+)/) {
            # BSD/macOS: m   ...
            ($id, $raw_key) = ($1, $2);
        }
        elsif ($line =~ /^\s*(\d+)\s+(0x[0-9a-fA-F]+)\s+/) {
            # DragonFly BSD:   ... (no type-letter column)
            ($id, $raw_key) = ($1, $2);
        }
        elsif ($line =~ /^\s*(\S+)\s+(\d+)\s+\S+/) {
            # Linux:   ...
            ($raw_key, $id) = ($1, $2);
        }
        else {
            next;
        }

        my $key_int = $raw_key =~ /^0x[0-9a-fA-F]+$/
            ? hex($raw_key)
            : $raw_key =~ /^-?\d+$/
            ? int($raw_key)
            : next;

        # IPC_PRIVATE segments cannot be re-attached across processes
        next if $key_int == 0;

        my $sem = IPC::Semaphore->new($key_int, 0, 0);
        next unless defined $sem;

        next unless _testing_semaphore_value($sem) == $target;

        # Don't tear down a segment another *live* process owns -- eg. a sibling
        # test file under `prove -j`, or a concurrent smoker testing the same
        # dist. Segments this process created ($cpid == $$), and orphans whose
        # creator process has exited, are still removed.

        my $probe = bless {}, 'IPC::Shareable::SharedMem';
        $probe->id($id);

        my $stat = eval { $probe->stat };
        my $cpid = defined $stat ? $stat->cpid : undef;

        next if defined $cpid && $cpid > 0 && $cpid != $$ && kill 0, $cpid;

        if (shmctl($id, IPC_RMID, 0)) {
            $sem->remove;
            delete $process_register{$id};
            delete $global_register{$id};
            $removed++;
        }
        else {
            warn "clean_up_testing(): could not remove shm segment $id: $!";
        }
    }
    close $ipcs_fh;

    # Second pass: reclaim orphaned testing-tagged semaphore sets -- ones whose
    # shm segment is already gone (eg. a crashed run died between removing the
    # segment and removing its semaphore set). The first pass cannot see these
    # because it walks ipcs -m. Each orphan pins a SEMMNI slot forever, and on
    # hosts with a tiny limit (OpenBSD defaults to kern.seminfo.semmni=10) the
    # accumulation eventually starves every subsequent semget() into ENOSPC --
    # the mass CPAN tester failure mode. A tagged set whose segment is gone can
    # never be re-attached by _tie() (the segment is always created before the
    # semaphore set), so removing it is race-free.

    open $ipcs_fh, '-|', 'ipcs', '-s' or die "ipcs -s: $!";

    while (my $line = <$ipcs_fh>) {
        my $raw_key;

        if ($line =~ /^\s*s\s+\d+\s+(\S+)/) {
            # BSD/macOS: s   ...
            $raw_key = $1;
        }
        elsif ($line =~ /^\s*\d+\s+(0x[0-9a-fA-F]+)\s+/) {
            # DragonFly BSD:   ... (no type-letter column)
            $raw_key = $1;
        }
        elsif ($line =~ /^\s*(\S+)\s+\d+\s+\S+/) {
            # Linux:   ...
            $raw_key = $1;
        }
        else {
            next;
        }

        my $key_int = $raw_key =~ /^0x[0-9a-fA-F]+$/
            ? hex($raw_key)
            : $raw_key =~ /^-?\d+$/
            ? int($raw_key)
            : next;

        next if $key_int == 0;

        # A live segment means a healthy (or first-pass handled) pair

        next if defined shmget($key_int, 0, 0);

        my $sem = IPC::Semaphore->new($key_int, 0, 0);
        next unless defined $sem;

        next unless _testing_semaphore_value($sem) == $target;

        $removed++ if $sem->remove;
    }
    close $ipcs_fh;

    return $removed;
}

# Private methods

# Encoding/Decoding

sub _encode {
    my ($knot, $seg, $data) = @_;

    # A scalar tie() holding a plain (defined, non-ref) value is stored verbatim
    # in a single segment — no serializer wrapping or escaping. Automatic and
    # serializer-agnostic; refs and undef fall through to the configured
    # serializer (refs fan out / freeze; undef → {"__sv__":null} / storable).

    if ($knot->{_type_int} == TYPE_SCALAR && ref($data) eq 'SCALAR') {
        my $val = $$data;

        if (defined $val && ! ref $val) {
            return _encode_verbatim($seg, $val);
        }
    }

    my $serializer = $knot->attributes('serializer');

    if ($serializer eq 'storable') {
        return _freeze($seg, $data);
    }

    return _encode_json($seg, $data);
}
sub _decode {
    my ($knot, $seg) = @_;

    # A scalar tie's value may have been stored verbatim (tag + \x1e sentinel);
    # recognize that and short-circuit before serializer dispatch, regardless of
    # json/storable.

    if ($knot->{_type_int} == TYPE_SCALAR) {
        my $verbatim = _decode_verbatim($seg);
        return $verbatim if defined $verbatim;
    }

    my $serializer = $knot->attributes('serializer');

    my $data = $serializer eq 'storable'
        ? _thaw($seg)
        : _decode_json($seg, $knot);

    return $data if defined $data;

    # Empty/never-written segment — return appropriate empty default so that
    # aggregate tie methods (FETCHSIZE, PUSH, CLEAR, etc.) can deref safely.

    return [] if $knot->{_type_int} == TYPE_ARRAY;
    return {} if $knot->{_type_int} == TYPE_HASH;

    return undef;
}
sub _encode_json {
    my $seg  = shift;
    my $data = shift;

    my $json = encode_json _encode_json_prepare($data);

    substr $json, 0, 0, 'IPC::Shareable';

    if (length($json) > $seg->size) {
        croak "Length of shared data exceeds shared segment size";
    }

    $seg->shmwrite($json);
}
sub _encode_json_prepare {
    my ($data) = @_;

    my $type = Scalar::Util::reftype($data) or return $data;

    # Replace direct IPC::Shareable child segments with __ics__ markers.

    # All nested refs are tied children — no recursion needed; each child
    # segment encodes its own children independently. We have to do this because
    # JSON can't store blessed objects

    if ($type eq 'HASH') {
        {
            my $has_child = 0;
            for my $val (values %$data) {
                if (ref($val) && _is_child($val)) {
                    $has_child = 1;
                    last;
                }
            }
            return $data if ! $has_child;
        }

        my %result;

        for my $key (keys %$data) {
            my $val   = $data->{$key};
            my $inner = ref($val) && _is_child($val);

            $result{$key} = $inner
                ? { '__ics__' => { type => $inner->{_type}, child_key => $inner->{_key}, child_key_hex => sprintf('0x%08x', $inner->{_key}) } }
                : $val;
        }

        return \%result;
    }

    if ($type eq 'ARRAY') {
        {
            my $has_child = 0;
            for my $val (@$data) {
                if (ref($val) && _is_child($val)) {
                    $has_child = 1;
                    last;
                }
            }
            return $data if ! $has_child;
        }

        return [
            map {
                my $inner = ref($_) && _is_child($_);
                $inner
                    ? { '__ics__' => { type => $inner->{_type}, child_key => $inner->{_key}, child_key_hex => sprintf('0x%08x', $inner->{_key}) } }
                    : $_
            } @$data
        ];
    }

    if ($type eq 'SCALAR' || $type eq 'REF') {
        my $val   = $$data;
        my $inner = ref($val) && _is_child($val);

        return $inner
            ? { '__ics__' => { type => $inner->{_type}, child_key => $inner->{_key}, child_key_hex => sprintf('0x%08x', $inner->{_key}) } }
            : { '__sv__' => $val };
    }

    return $data;
}
sub _decode_json {
    my ($seg, $knot) = @_;

    my $json = $seg->data;

    return if ! $json;

    # The return of shmread() is the actual size of the defined size of the
    # shared memory segment. Even if the return equates to an empty string
    # (which it will if it contains no data), there will always be a length().
    # Therefore, we must see if we've tagged this data as a valid structure,
    # or else decode will fail

    my $tag = substr $json, 0, 14, '';

    if ($tag eq 'IPC::Shareable') {
        my $data = decode_json $json;

        if (! defined($data)) {
            croak "Munged shared memory segment (size exceeded?)";
        }

        if (defined $knot && index($json, '"__ics__"') >= 0) {
            _decode_json_restore($data, $knot)
        }

        # Unwrap scalar-tie values encoded as { '__sv__' => val } or
        # { '__ics__' => {...} }

        if (defined $knot && $knot->{_type_int} == TYPE_SCALAR && ref($data) eq 'HASH') {
            if (exists $data->{'__ics__'}) {
                my $prev     = $knot->{_data};
                my $prev_val = (defined $prev && ref($prev)) ? $$prev : undef;
                my $resolved = _decode_json_resolve($data->{'__ics__'}, $prev_val, $knot);
                return \$resolved;
            }
            if (exists $data->{'__sv__'}) {
                my $val = $data->{'__sv__'};
                return \$val;
            }
        }

        return $data;
    }
    else {
        return;
    }
}
sub _decode_json_restore {
    my ($data, $knot) = @_;

    my $type = Scalar::Util::reftype($data) or return;

    # Reuse existing tied child refs from previous decode where possible.
    # This avoids a shmget+semget system call pair for each child on every
    # decode cycle — only the first attach per segment incurs that cost.

    my $prev = $knot->{_data};

    if ($type eq 'HASH') {
        my $prev_is_hash = ref($prev) eq 'HASH';

        for my $key (keys %$data) {
            next unless ref($data->{$key}) eq 'HASH' && exists $data->{$key}{'__ics__'};

            $data->{$key} = _decode_json_resolve(
                $data->{$key}{'__ics__'},
                $prev_is_hash ? $prev->{$key} : undef,
                $knot,
            );
        }
    }
    elsif ($type eq 'ARRAY') {
        my $prev_is_array = ref($prev) eq 'ARRAY';
        my $prev_max = $prev_is_array ? $#$prev : -1;

        for my $i (0 .. $#$data) {
            next unless ref($data->[$i]) eq 'HASH' && exists $data->[$i]{'__ics__'};

            $data->[$i] = _decode_json_resolve(
                $data->[$i]{'__ics__'},
                $prev_is_array && $i <= $prev_max ? $prev->[$i] : undef,
                $knot,
            );
        }
    }
}
sub _decode_json_resolve {
    my ($info, $existing, $knot) = @_;

    if (defined $existing) {
        my $inner = ref($existing) && _is_child($existing);
        return $existing if $inner && $inner->{_key} == $info->{child_key};
    }

    return _decode_json_reattach($info, $knot);
}
sub _decode_json_reattach {
    my ($info, $knot) = @_;

    my %opts = (
        %{ $knot->attributes },
        key       => $info->{child_key},
        exclusive => 0,
        create    => 0,
        magic     => 1,
    );

    if ($info->{type} eq 'HASH') {
        my %h;
        tie %h, 'IPC::Shareable', \%opts;
        return \%h;
    }
    elsif ($info->{type} eq 'ARRAY') {
        my @a;
        tie @a, 'IPC::Shareable', \%opts;
        return \@a;
    }
    elsif ($info->{type} eq 'SCALAR') {
        my $s;
        tie $s, 'IPC::Shareable', \%opts;
        return \$s;
    }
}
sub _encode_verbatim {
    my ($seg, $val) = @_;

    # Store a plain scalar verbatim. Layout: the 14-byte 'IPC::Shareable' tag
    # (so shm_segments()/clean_up_testing still recognize the segment as ours),
    # a one-byte \x1e sentinel marking "not serialized — hand these bytes back
    # as-is", then the caller's bytes. The sentinel lets _decode tell this from
    # a json {…} body or a storable header. The caller (_encode) guarantees
    # $val is a defined, non-ref scalar.

    my $raw = "IPC::Shareable\x1e" . $val;

    if (length($raw) > $seg->size) {
        croak "Length of shared data exceeds shared segment size";
    }

    $seg->shmwrite($raw);
}
sub _decode_verbatim {
    my ($seg) = @_;

    # Recognize a verbatim scalar segment: the 14-byte 'IPC::Shareable' tag
    # followed by the \x1e sentinel. Return a scalar ref to the bytes after the
    # sentinel — trailing NUL padding stripped, internal NULs preserved. Return
    # undef for anything else (a json {…}/[…] body, a storable body, or an
    # empty/never-written segment) so _decode falls through to the serializer.

    my $raw = $seg->shmread;

    return if ! defined $raw;

    $raw =~ s/\x00+$//;

    return if substr($raw, 0, 15) ne "IPC::Shareable\x1e";

    my $payload = substr($raw, 15);

    return \$payload;
}
sub _freeze {
    my ($seg, $water) = @_;

    my $ice = freeze $water;

    croak "Could not serialize data for shared memory"
        unless defined $ice;

    substr $ice, 0, 0, 'IPC::Shareable';

    if (length($ice) > $seg->size) {
        croak "Length of shared data exceeds shared segment size";
    }

    $seg->shmwrite($ice);
}
sub _thaw {
    my ($seg) = @_;

    my $ice = $seg->shmread;

    return if ! $ice;

    my $tag = substr $ice, 0, 14, '';

    if ($tag eq 'IPC::Shareable') {
        my $water = thaw $ice;
        if (! defined($water)) {
            croak "Munged shared memory segment (size exceeded?)";
        }
        return $water;
    }
    else {
        return;
    }
}

# Data management

sub _tie {
    my ($type, $class, $key_str, $opts);

    if (scalar @_ == 4) {
        # Legacy API allowed a string scalar key
        ($type, $class, $key_str, $opts) = @_;
        $opts->{key} = $key_str;
    }
    else {
        ($type, $class, $opts) = @_;
    }

    $opts  = _parse_args($opts);

    my $knot = bless { attributes => $opts }, $class;

    $knot->uuid;

    my $key      = $knot->_shm_key;
    my $flags    = $knot->_shm_flags;
    my $shm_size = $knot->attributes('size');

    if ($knot->attributes('limit') && $shm_size > SHMMAX_BYTES) {
        croak
            "Shared memory segment size '$shm_size' is larger than max size of " .
            SHMMAX_BYTES;
    }

    my $seg;

    if ($knot->attributes('graceful')) {
        my $exclusive = eval {
            $seg = IPC::Shareable::SharedMem->new(
                key   => $key,
                size  => $shm_size,
                flags => $flags,
                mode  => $knot->attributes('mode'),
                type  => $type,
            );

            1;
        };

        if (! defined $exclusive) {
            if ($knot->attributes('warn')) {
                my $key = lc(sprintf("0x%X", $knot->_shm_key));

                warn "Process ID $$ exited due to exclusive shared memory collision at segment/semaphore key '$key'\n";
            }
            exit(0);
        }
    }
    else {
        $seg = IPC::Shareable::SharedMem->new(
            key   => $key,
            size  => $shm_size,
            flags => $flags,
            mode  => $knot->attributes('mode'),
            type  => $type,
        );
    }

    if (! defined $seg) {
        if ($!{ENOMEM}) {
            croak "\nERROR: Could not create shared memory segment: $!\n\n" .
                  "Are you using too large a segment size, or spawning too many segments?";
        }

        if ($!{ENOSPC}) {
            croak "\nERROR: Could not create shared memory segment: $!\n\n" .
                "Are you spawning too many segments (in a loop perhaps)?";
        }

        if (! $knot->attributes('create')) {
            confess "ERROR: Could not acquire shared memory segment... 'create' ".
                  "option is not set, and the segment hasn't been created " .
                  "yet:\n\n $!";
        }
        elsif ($knot->attributes('create') && $knot->attributes('exclusive')) {
            croak "ERROR: Could not create shared memory segment. 'create' " .
                  "and 'exclusive' are set. Does the segment already exist? " .
                  "\n\n$!";
        }
        else {
            croak "ERROR: Could not create shared memory segment.\n\n$!";
        }
    }

    # Try to attach to an existing semaphore set first using nsems=0, which
    # avoids EINVAL on macOS/BSD when the existing set has fewer slots than
    # the requested count. If the set does not exist yet, fall through to
    # create a new semaphore set: 5 slots when the 'testing' attribute is set
    # (adds SEM_TESTING at index 4), 4 slots otherwise.

    my $nsems = $knot->attributes('testing') ? 5 : 4;

    my $sem = IPC::Semaphore->new($key, 0, $seg->flags & 0777)
           // IPC::Semaphore->new($key, $nsems, $seg->flags);

    if (! defined $sem) {
        # The segment was created just above, but we couldn't establish its
        # semaphore set (eg. ENOSPC when the host's semaphore limit is hit).
        # Remove the segment we just made so it isn't orphaned -- but only when
        # we are the creator: a pure attacher (create => 0) must never remove a
        # segment that another process owns. An IPC_PRIVATE segment is always
        # freshly created by shmget() regardless of the 'create' attribute, and
        # is unreachable by key once we croak, so it must be removed too -- it
        # would otherwise leak invisibly (clean_up_testing() cannot see key 0).
        # Preserve $! across the removal so the croak still reports the
        # original failure (eg. "No space left on device") rather than the
        # result of the cleanup's shmctl.

        my $err = $!;

        $seg->remove if $knot->attributes('create') || $key == IPC_PRIVATE;
        $! = $err;

        croak "Could not create semaphore set: $!\n";
    }

    if (! $sem->op(@{ $semop_args{(LOCK_SH)} }) ) {
        # Lock acquisition failed before the knot was registered, so nothing
        # else will reclaim these. Tear down what we just made: the semaphore
        # set if we created it (its marker isn't set yet), and the segment if
        # we are its creator (an IPC_PRIVATE segment is always freshly created,
        # and unreachable by key hereafter, so it counts as ours too).
        # Preserve $! so the croak still names the cause.

        my $err = $!;

        $sem->remove if $sem->getval(SEM_MARKER) != SHM_EXISTS;
        $seg->remove if $knot->attributes('create') || $key == IPC_PRIVATE;
        $! = $err;

        croak "Could not obtain semaphore set lock: $!\n";
    }

    %$knot = (
        %$knot,
        _hkey_list          => undef,
        _key                => $key,
        _key_hex            => $seg->key_hex,
        _lock               => 0,
        _shm                => $seg,
        _sem                => $sem,
        _type               => $type,
        _type_int           => $type eq 'HASH' ? TYPE_HASH : $type eq 'ARRAY' ? TYPE_ARRAY : TYPE_SCALAR,
        _was_changed        => 0,
    );

    my $serializer = $knot->attributes('serializer');

    if ($serializer eq 'json') {
        my $data;
        my $decoded_ok = eval { $data = $knot->_decode($seg); 1 };

        if (! $decoded_ok) {
            # JSON decode threw; the segment may contain legacy Storable data.
            # Try Storable; if it succeeds, silently switch this session over
            # and warn the caller so they know to migrate.

            my $storable_data;
            my $thaw_ok = eval { $storable_data = _thaw($seg); 1 };

            if ($thaw_ok && defined $storable_data) {
                carp sprintf(
                    "IPC::Shareable: segment 0x%08x contains Storable-encoded data; "
                  . "switching serializer to 'storable' for this session. "
                  . "Re-create the segment to migrate it to JSON.",
                    $key
                );
                $knot->{attributes}{serializer} = 'storable';
                $knot->{_data} = $storable_data;
            }
            else {
                die $@;
            }
        }
        else {
            $knot->{_data} = $data;
        }
    }
    else {
        $knot->{_data} = $knot->_decode($seg);
    }

    # Register unconditionally so any process that attaches to an existing
    # segment (create=>0, re-attach, cross-process) is also tracked for
    # clean_up_all(). Previously only new segments were registered here,
    # requiring the Dumper hack in global_register() to catch the rest

    if (! exists $global_register{$knot->seg->id}) {
        $global_register{$knot->seg->id} = $knot;
    }

    if ($sem->getval(SEM_MARKER) != SHM_EXISTS) {

        $process_register{$knot->seg->id} ||= $knot;

        $sem->setval(SEM_PROTECTED, $knot->attributes('protected'));

        if ($knot->attributes('testing')) {
            $sem->setval(SEM_TESTING, _testing_semaphore_key_hash($knot->attributes('testing')));
        }

        if (! $sem->setval(SEM_MARKER, SHM_EXISTS)) {
            croak "Couldn't set semaphore during object creation: $!";
        }
    }
    else {
        # Segment already existed — restore the protected and testing
        # attributes from the semaphore so that clean_up_all() / clean_up_testing()
        # in this process work correctly even when the caller did not explicitly
        # pass them on tie.

        my $stored_protected = $sem->getval(SEM_PROTECTED);

        if (defined $stored_protected && $stored_protected != 0) {
            $knot->{attributes}{protected} = $stored_protected
        }

        my $stored_testing = _testing_semaphore_value($sem);

        if ($stored_testing) {
            $knot->{attributes}{testing} = $stored_testing;
        }
    }

    $sem->op(@{ $semop_args{(LOCK_SH|LOCK_UN)} });

    return $knot;
}
sub _magic_tie {
    my ($parent, $val) = @_;

    my $key;

    if ($parent->{_key} == IPC_PRIVATE && $parent->attributes('serializer') ne 'json') {
        $key = IPC_PRIVATE;
    }
    else {
        $key = _shm_key_rand();
    }

    # The individual options in the hash override any pre-set options that are
    # being inherited from the parent

    my %opts = (
        %{ $parent->attributes },
        key       => $key,
        exclusive => 1,
        create    => 1,
        magic     => 1,
    );

    # XXX I wish I didn't have to take a copy of data here and copy it back in
    # XXX Also, have to peek inside potential objects to see their implementation

    my $child;
    my $type = Scalar::Util::reftype($val) || '';

    if ($type eq "HASH") {
        my %copy = %$val;
        $child = tie %$val, 'IPC::Shareable', $key, { %opts };
        croak "Could not create inner tie" if ! $child;

        %$val = %copy;
    }
    elsif ($type eq "ARRAY") {
        my @copy = @$val;
        $child = tie @$val, 'IPC::Shareable', $key, { %opts };
        croak "Could not create inner tie" if ! $child;

        @$val = @copy;
    }
    elsif ($type eq "SCALAR") {
        my $copy = $$val;
        $child = tie $$val, 'IPC::Shareable', $key, { %opts };
        croak "Could not create inner tie" if ! $child;

        $$val = $copy;
    }
    else {
        croak "Variables of type $type not implemented";
    }

    return $child;
}
sub _need_tie {
    my ($knot, $val) = @_;

    my $type = Scalar::Util::reftype($val);
    return 0 if ! $type;

    my $need_tie;

    if ($type eq "HASH") {
        $need_tie = ! (tied %$val);
    }
    elsif ($type eq "ARRAY") {
        $need_tie = ! (tied @$val);
    }
    elsif ($type eq "SCALAR") {
        $need_tie = ! (tied $$val);
    }

    return $need_tie ? 1 : 0;
}
sub _remove_child {
    my ($val) = @_;

    if (ref($val) && (my $child = _is_child($val))) {
        $child->remove;
    }
}
sub _is_child {
    return $_have_xs
        ? _is_child_xs($_[0])
        : _is_child_pp($_[0]);
}
sub _is_child_pp {
    my $data = shift or return;

    my $type = Scalar::Util::reftype( $data );
    return unless $type;

    my $obj;

    if ($type eq "HASH") {
        $obj = tied %$data;
    }
    elsif ($type eq "ARRAY") {
        $obj = tied @$data;
    }
    elsif ($type eq "SCALAR") {
        $obj = tied $$data;
    }

    if (ref $obj eq 'IPC::Shareable') {
        return $obj;
    }

    return;
}
sub _write_to_seg {
    my ($knot) = @_;

    my $seg_id = $knot->seg->id;

    if (! defined $knot->_encode($knot->seg, $knot->{_data})) {
        croak "Could not write to shared memory segment $seg_id: $!";
    }
}

# Segment/semaphore operations

sub _execute_lock_coderef {
    my ($knot, $code) = @_;

    my $ok = eval { $code->(); 1 };
    my $err = $@;

    $knot->unlock;

    die $err if ! $ok;
}
sub _key_str_to_int {
    # Convert any key format (hex string, decimal integer string, or arbitrary
    # text) to a 32-bit integer using the same algorithm as _shm_key(), but
    # without the %used_ids side effect. Safe to call any number of times.

    my ($key_str) = @_;

    return hex($key_str)    if $key_str =~ /^0x[0-9a-fA-F]+$/i;
    return $key_str + 0     if $key_str =~ /^\d+$/;

    my $int = crc32($key_str);
    $int -= MAX_KEY_INT_SIZE if $int > MAX_KEY_INT_SIZE;
    return $int;
}
sub _lock_children {
    my ($root_knot, $flags) = @_;

    my @locked;

    my %seen = ($root_knot->seg->id => 1);
    my @stack = ([$root_knot, 0]);

    while (@stack) {
        my $frame = $stack[-1];
        my ($knot, $idx) = @$frame;

        my $data  = $knot->{_data};
        my $rtype = Scalar::Util::reftype($data) // '';

        my @vals = $rtype eq 'HASH'  ? values %$data
                 : $rtype eq 'ARRAY' ? @$data
                 : ();

        my $found = 0;

        for (my $i = $idx; $i < @vals; $i++) {

            my $val = $vals[$i];
            next unless ref($val);

            my $child = _is_child($val);
            next unless $child && $child->seg;

            my $id = $child->seg->id;
            next if $seen{$id}++;

            if (! $child->sem->op(@{ $semop_args{$flags} })) {
                for my $locked (reverse @locked) {
                    my $rflags = $locked->{_lock} | LOCK_UN;
                    $rflags ^= LOCK_NB if $rflags & LOCK_NB;
                    $locked->sem->op(@{ $semop_args{$rflags} });
                    $locked->{_lock} = 0;
                }
                return;
            }

            $child->{_data} = $child->_decode($child->seg);
            $child->{_lock} = $flags;

            push @locked, $child;

            $frame->[1] = $i + 1;

            push @stack, [$child, 0];

            $found = 1;

            last;
        }

        pop @stack unless $found;
    }

    return \@locked;
}
sub _shm_data_summary {
    my ($knot) = @_;

    my $data  = $knot->{_data};
    my $rtype = Scalar::Util::reftype($data) // '';

    if ($rtype eq 'SCALAR') {
        my $v = $$data;
        return defined $v ? qq("$v") : '(undef)';
    }

    if ($rtype eq 'HASH') {
        my @parts;
        for my $k (sort keys %$data) {
            my $v = $data->{$k};
            if (ref $v) {
                my $vt    = Scalar::Util::reftype($v) // '';
                my $child = $vt eq 'HASH'   ? tied(%$v)
                    : $vt eq 'ARRAY'  ? tied(@$v)
                    : $vt eq 'SCALAR' ? tied($$v)
                    : undef;
                push @parts, $child && $child->{_key_hex}
                    ? qq($k => {_key_hex}>)
                    : "$k => ";
            }
            else {
                push @parts, defined $v ? qq($k => "$v") : "$k => (undef)";
            }
        }
        return @parts ? '{ ' . join(', ', @parts) . ' }' : '{}';
    }

    if ($rtype eq 'ARRAY') {
        my @parts;
        for my $v (@$data) {
            if (ref $v) {
                my $vt    = Scalar::Util::reftype($v) // '';
                my $child = $vt eq 'HASH'   ? tied(%$v)
                    : $vt eq 'ARRAY'  ? tied(@$v)
                    : $vt eq 'SCALAR' ? tied($$v)
                    : undef;
                push @parts, $child && $child->{_key_hex}
                    ? "{_key_hex}>"
                    : '';
            }
            else {
                push @parts, defined $v ? qq("$v") : '(undef)';
            }
        }
        return '[' . join(', ', @parts) . ']';
    }

    return '(unknown type)';
}
sub _shm_flags {
    # Parses the anonymous hash passed to constructors; returns a list
    # of args suitable for passing to shmget

    my ($knot) = @_;

    my $flags = 0;

    $flags |= IPC_CREAT if $knot->attributes('create');
    $flags |= IPC_EXCL  if $knot->attributes('exclusive');

    return $flags;
}
sub _shm_key {
    # Generates a 32-bit CRC on the key string. The $key_str parameter is used
    # for testing only, for purposes of testing various key strings

    my ($knot, $key_str) = @_;

    $key_str //= ($knot->attributes('key') || '');

    my $key;

    if ($key_str eq '') {
        $key = IPC_PRIVATE;
    }
    elsif ($key_str =~ /^0x[0-9a-fA-F]+$/i) {
        # User specified an explicit hex string key (eg. '0xDEADBEEF'); use the

        # bit pattern as-is so the segment key seen by ipcs(1) matches exactly.
        $key = hex($key_str);
        $used_ids{$key}++;
        return $key;
    }
    elsif ($key_str =~ /^\d+$/) {
        # User specified an explicit decimal integer key; use it as-is.
        $key = $key_str;
        $used_ids{$key}++;
        return $key;
    }
    else {
        # String key: compute a 32-bit CRC and apply overflow correction so the
        # result fits in a signed 32-bit key_t.
        $key = crc32($key_str);
    }

    $used_ids{$key}++;

    if ($key >= MAX_KEY_INT_SIZE) {
        $key = $key - MAX_KEY_INT_SIZE;

        if ($key == 0) {
            croak "We've calculated a key which equals 0. This is a fatal error";
        }
    }

    return $key;
}
sub _shm_key_rand {
    my $key;

    # Unfortunately, the only way I know how to check if a segment exists is
    # to actually create it. We must do that here, then remove it just to
    # ensure the slot is available

    my $verified_exclusive = 0;

    my $check_count = 0;

    while (! $verified_exclusive && $check_count < EXCLUSIVE_CHECK_LIMIT) {
        $check_count++;

        $key = _shm_key_rand_int();

        next if $used_ids{$key};

        my $flags;
        $flags |= IPC_CREAT;
        $flags |= IPC_EXCL;

        my $seg;

        my $shm_slot_available = eval {
            $seg = IPC::Shareable::SharedMem->new(
                key     => $key,
                size    => 1,
                flags   => $flags,
            );

            1;
        };

        if ($shm_slot_available) {
            $verified_exclusive = 1;
            $seg->remove if $seg;
        }
    }

    if (! $verified_exclusive) {
        croak
            "_shm_key_rand() can't get an available key after $check_count tries";
    }

    $used_ids{$key}++;

    return $key;
}
sub _shm_key_rand_int {
    return int(rand(1_000_000));
}
sub _read_check {
    my ($knot) = @_;

    # Advisory only: never blocks the read, only warns. Called from FETCH
    # when this knot is unlocked (a locked FETCH uses _data cache and never
    # touches shmem). Race window exists between this getval() and the
    # subsequent _decode() — a writer could acquire in between — but this
    # still catches the common case where a reader forgot to lock.

    return unless $knot->attributes('enforced_read_locking');
    return unless $knot->attributes('violated_read_lock_warn');

    # getval() can return undef if the semaphore set has been removed (eg.
    # after clean_up_all). The check is advisory only, so silently skip when
    # the semaphore is no longer reachable.

    my $writers = $knot->sem->getval(SEM_WRITERS);
    return unless defined $writers;

    if ($writers > 0) {
        my $uuid   = $knot->uuid;
        my $seg_id = $knot->seg->id;

        warn "Object with UUID $uuid attempted read from segment ID "
            . "$seg_id which is exclusively locked (enforced read locking "
            . "enabled); returned data may be stale or partially-written. "
            . "Acquire LOCK_SH before reading to guarantee a coherent snapshot";
    }

    return;
}
sub _write_permitted {
    my ($knot) = @_;

    return 1 unless $knot->attributes('enforced_write_locking');

    # If this knot itself holds LOCK_EX it is the owner of the lock and is
    # permitted to write.

    return 1 if $knot->{_lock} & LOCK_EX;

    my $sem = $knot->sem;

    # Semaphore index 2 is the write-lock counter; it is 1 when any other knot
    # holds LOCK_EX (set via SEM_UNDO so it auto-releases on process exit).

    # getval() returns undef if the semaphore set has been removed by another
    # process (eg. clean_up_all, or a peer with destroy=>1 exiting). The
    # enforcement check is advisory, so when the set is unreachable we skip it
    # and permit the write, mirroring _check_read_lock().

    # Block if any process holds LOCK_EX

    my $writers = $sem->getval(SEM_WRITERS);
    return 1 if ! defined $writers;

    if ($writers > 0) {
        if ($knot->attributes('violated_write_lock_warn')) {
            my $uuid   = $knot->uuid;
            my $seg_id = $knot->seg->id;

            warn "Object with UUID $uuid attempted write to segment ID "
                . "$seg_id which is exclusively locked (enforced write "
                . "locking enabled). Your write was not accepted. Lock with "
                . "LOCK_EX to ensure successful writes when a segment is "
                . "already locked";
        }

        return 0;
    }

    # Block if any process holds LOCK_SH (active readers present)

    my $readers = $sem->getval(SEM_READERS);
    return 1 if ! defined $readers;

    if ($readers > 0) {
        if ($knot->attributes('violated_write_lock_warn')) {
            my $uuid   = $knot->uuid;
            my $seg_id = $knot->seg->id;

            warn "Object with UUID $uuid attempted write to segment ID "
                . "$seg_id which has active readers (enforced write locking "
                . "enabled)";
        }

        return 0;
    }

    return 1;
}

# Unit testing support

sub _testing_semaphore_key_hash {
    my ($dist_name) = @_;

    # SysV SEMVMX caps semaphore values at 32767 on most platforms (incl.
    # macOS, BSD); mask the CRC32 to 15 bits so setval() never silently fails.
    # 0 is reserved to mean "not a testing segment", so we shift any zero
    # collision off slot 0.

    my $h = String::CRC32::crc32($dist_name) & 0x7FFF;

    return $h || 1;
}
sub _testing_semaphore_value {
    my ($sem) = @_;

    my $stat = $sem->stat or return 0;

    return 0 if $stat->nsems < 5;
    return $sem->getval(SEM_TESTING) // 0;
}

# Misc

sub _parse_args {
    my ($opts) = @_;

    $opts  = defined $opts  ? $opts  : { %default_options };

    # Note caller's explicit intent BEFORE defaults are merged in. A caller
    # who passes testing => 0 wants to opt out of auto-tagging; we must not
    # treat that as "absent" after defaulting.

    my $testing_explicit = exists $opts->{testing};

    for my $k (keys %default_options) {
        if (not defined $opts->{$k}) {
            $opts->{$k} = $default_options{$k};
        }
        elsif ($opts->{$k} eq 'no') {
            if ($^W) {
                require Carp;
                Carp::carp("Use of `no' in IPC::Shareable args is obsolete");
            }

            $opts->{$k} = 0;
        }
    }

    # Validate the serializer selection. 'json' (default) and 'storable' are the
    # only user-selectable options

    my $serializer = $opts->{serializer};

    if ($serializer ne 'json' && $serializer ne 'storable') {
        croak "Invalid 'serializer' value '$serializer'; must be 'json' or 'storable'";
    }

    $opts->{owner} = ($opts->{owner} or $$);
    $opts->{magic} = ($opts->{magic} or 0);

    # Inherit the process-level testing tag set by testing_set(), unless the
    # caller explicitly passed a testing value (including testing => 0)

    if ($_testing_dist && ! $testing_explicit) {
        $opts->{testing} = $_testing_dist;
    }

    return $opts;
}
sub _end {
    for my $s (values %process_register) {
        eval { unlock($s) };
        next if $s->attributes('protected');
        next if ! $s->attributes('destroy');
        next if $s->attributes('owner') != $$;
        eval { remove($s) };
    }
}

END {
    _end();
}

sub _placeholder {}

1;

__END__