File Coverage

blib/lib/Atomic/Pipe.pm

Criterion	Covered	Total	%
statement	623	686	90.8
branch	251	346	72.5
condition	183	288	63.5
subroutine	98	103	95.1
pod	41	45	91.1
total	1196	1468	81.4

package Atomic::Pipe;
use strict;
use warnings;

our $VERSION = '0.031';

use IO();
use IO::Handle();
use Fcntl();
use bytes();

BEGIN {
    if (eval { require IO::Select; 1 }) {
        *HAVE_IO_SELECT = sub() { 1 };
    }
    else {
        *HAVE_IO_SELECT = sub() { 0 };
    }
}

use Carp qw/croak confess/;
use Config qw/%Config/;
use List::Util qw/min/;
use Scalar::Util qw/blessed/;

use Errno qw/EINTR EAGAIN EPIPE/;
my (%RETRY_ERRNO, %NONBLOCK_ERRNO);
BEGIN {
    %RETRY_ERRNO = (EINTR() => 1);
    $RETRY_ERRNO{Errno->ERESTART} = 1 if Errno->can('ERESTART');

    # EWOULDBLOCK == EAGAIN on most platforms, but POSIX allows them to differ.
    %NONBLOCK_ERRNO = (EAGAIN() => 1);
    $NONBLOCK_ERRNO{Errno->EWOULDBLOCK} = 1 if Errno->can('EWOULDBLOCK');
}

BEGIN {
    # POSIX says writes of 512 or less are atomic, but some platforms allow for
    # larger ones.
    require POSIX;
    if (POSIX->can('PIPE_BUF') && eval { POSIX::PIPE_BUF() }) {
        *PIPE_BUF = \&POSIX::PIPE_BUF;
    }
    else {
        *PIPE_BUF = sub() { 512 };
    }

    if (POSIX->can('SSIZE_MAX') && eval { POSIX::SSIZE_MAX() }) {
        *SSIZE_MAX = \&POSIX::SSIZE_MAX;
    }
    else {
        # POSIX guarantees SSIZE_MAX is at least 32767 (_POSIX_SSIZE_MAX).
        *SSIZE_MAX = sub() { 32767 };
    }

    {
        # Using the default pipe size as a read size is significantly faster
        # than a larger value on my test machine.
        my $read_size = min(SSIZE_MAX(), 65_536);
        *DEFAULT_READ_SIZE = sub() { $read_size };
    }

    my $can_thread = 1;
    $can_thread &&= $] >= 5.008001;
    $can_thread &&= $Config{'useithreads'};

    # Threads are broken on perl 5.10.0 built with gcc 4.8+
    if ($can_thread && $] == 5.010000 && $Config{'ccname'} eq 'gcc' && $Config{'gccversion'}) {
        my @parts = split /\./, $Config{'gccversion'};
        $can_thread = 0 if $parts[0] > 4 || ($parts[0] == 4 && $parts[1] >= 8);
    }

    $can_thread &&= !$INC{'Devel/Cover.pm'};

    if (!$can_thread) {
        *_get_tid = sub() { 0 };
    }
    elsif ($INC{'threads.pm'}) {
        *_get_tid = sub() { threads->tid() };
    }
    else {
        *_get_tid = sub() { $INC{'threads.pm'} ? threads->tid() : 0 };
    }

    if ($^O eq 'MSWin32') {
        local $@;
        eval { require Win32::API;     1 } or die "non-blocking on windows requires Win32::API please install it.\n$@";
        eval { require Win32API::File; 1 } or die "non-blocking on windows requires Win32API::File please install it.\n$@";
        *IS_WIN32 = sub() { 1 };
    }
    else {
        *IS_WIN32 = sub() { 0 };
    }
}

use constant READ_SIZE      => 'read_size';
use constant RH             => 'rh';
use constant WH             => 'wh';
use constant EOF            => 'eof';
use constant STATE          => 'state';
use constant OUT_BUFFER     => 'out_buffer';
use constant IN_BUFFER      => 'in_buffer';
use constant IN_BUFFER_SIZE => 'in_buffer_size';
use constant READ_BLOCKING  => 'read_blocking';
use constant WRITE_BLOCKING => 'write_blocking';
use constant BURST_PREFIX   => 'burst_prefix';
use constant BURST_POSTFIX  => 'burst_postfix';
use constant ADJUSTED_DSIZE => 'adjusted_dsize';
use constant MESSAGE_KEY    => 'message_key';
use constant MIXED_BUFFER   => 'mixed_buffer';
use constant DELIMITER_SIZE => 'delimiter_size';
use constant INVALID_STATE  => 'invalid_state';
use constant HIT_EPIPE      => 'hit_epipe';
use constant USE_IO_SELECT  => 'use_io_select';
use constant COMPRESSION                 => 'compression';
use constant COMPRESSION_LEVEL           => 'compression_level';
use constant COMPRESSION_DICTIONARY      => 'compression_dictionary';
use constant COMPRESSION_DICTIONARY_FILE => 'compression_dictionary_file';
use constant KEEP_COMPRESSED             => 'keep_compressed';

sub wh  { shift->{+WH} }
sub rh  { shift->{+RH} }

sub throw_invalid {
    my $self = shift;
    $self->{+INVALID_STATE} //= @_ ? shift : 'Unknown Error';
    confess "Pipe is in an invalid state '$self->{+INVALID_STATE}'";
}

sub read_size {
    my $self = shift;
    ($self->{+READ_SIZE}) = @_ if @_;
    return $self->{+READ_SIZE} ||= DEFAULT_READ_SIZE();
}

sub use_io_select {
    my $self = shift;
    if (@_) {
        croak "IO::Select is not installed, cannot enable use_io_select" if $_[0] && !HAVE_IO_SELECT;
        $self->{+USE_IO_SELECT} = $_[0] ? 1 : 0;
        delete $self->{_select} unless $_[0];
    }
    return 0 unless HAVE_IO_SELECT;
    my $val = $self->{+USE_IO_SELECT};
    return defined($val) ? ($val ? 1 : 0) : IS_WIN32 ? 0 : 1;
}

sub compression                 { $_[0]->{+COMPRESSION} }
sub compression_level           { $_[0]->{+COMPRESSION_LEVEL} }
sub compression_dictionary      { $_[0]->{+COMPRESSION_DICTIONARY} }
sub compression_dictionary_file { $_[0]->{+COMPRESSION_DICTIONARY_FILE} }
sub keep_compressed             { $_[0]->{+KEEP_COMPRESSED} ? 1 : 0 }

sub fill_buffer {
    my $self = shift;

    $self->throw_invalid() if $self->{+INVALID_STATE};

    my $rh = $self->{+RH} or die "Not a read handle";

    return 0 if $self->{+EOF};

    $self->{+IN_BUFFER_SIZE} //= 0;

    my $to_read = $self->{+READ_SIZE} || DEFAULT_READ_SIZE();

    my $use_select = $self->use_io_select;

    if ($use_select) {
        my $sel = $self->{_select} //= IO::Select->new($rh);
        my $blocking = $self->{+READ_BLOCKING} // 1;
        my @ready = $sel->can_read($blocking ? undef : 0);
        return 0 unless @ready;
    }
    elsif (IS_WIN32 && defined($self->{+READ_BLOCKING}) && !$self->{+READ_BLOCKING}) {
        $to_read = min($self->_win32_pipe_ready(), $to_read);
    }

    return 0 unless $to_read;

    while (1) {
        my $rbuff = '';
        my $got = sysread($rh, $rbuff, $to_read);
        unless(defined $got) {
            return 0 if $NONBLOCK_ERRNO{0 + $!}; # NON-BLOCKING
            if ($RETRY_ERRNO{0 + $!}) {
                next unless $use_select; # retry on EINTR in fallback mode
                return 0;               # IO::Select handles EINTR
            }
            $self->throw_invalid("$!");
        }

        if ($got) {
            $self->{+IN_BUFFER} .= $rbuff;
            $self->{+IN_BUFFER_SIZE} += $got;
            return $got;
        }
        else {
            $self->{+EOF} = 1;
            return 0;
        }
    }

    return 0;
}

sub _get_from_buffer  { $_[0]->_from_buffer($_[1], remove => 1) }
sub _peek_from_buffer { shift->_from_buffer(@_) }

sub _from_buffer {
    my $self = shift;
    my ($size, %params) = @_;

    unless ($self->{+IN_BUFFER_SIZE} && $self->{+IN_BUFFER_SIZE} >= $size) {
        $self->fill_buffer;
        unless($self->{+IN_BUFFER_SIZE} >= $size) {
            return unless $params{eof_invalid} && $self->{+EOF};
            $self->throw_invalid($params{eof_invalid});
        }
    }

    my $out;

    if ($params{remove}) {
        $self->{+IN_BUFFER_SIZE} -= $size;
        $out = substr($self->{+IN_BUFFER}, 0, $size, '');
    }
    else {
        $out = substr($self->{+IN_BUFFER}, 0, $size);
    }

    return $out;
}

sub _has_dict {
    return defined($_[0]->{+COMPRESSION_DICTIONARY})
        || defined($_[0]->{+COMPRESSION_DICTIONARY_FILE});
}

# NOTE: raw zstd dictionaries do not embed a dict-ID, so a mismatched peer
# dict will silently decode to garbage rather than fail. Both ends must agree
# on byte-identical dictionary content.
sub _build_cdict {
    my $self = shift;
    my $level = $self->{+COMPRESSION_LEVEL} // 3;
    require Compress::Zstd::CompressionDictionary;
    if (defined(my $path = $self->{+COMPRESSION_DICTIONARY_FILE})) {
        return Compress::Zstd::CompressionDictionary->new_from_file($path, $level);
    }
    return Compress::Zstd::CompressionDictionary->new($self->{+COMPRESSION_DICTIONARY}, $level);
}

sub _build_ddict {
    my $self = shift;
    require Compress::Zstd::DecompressionDictionary;
    if (defined(my $path = $self->{+COMPRESSION_DICTIONARY_FILE})) {
        return Compress::Zstd::DecompressionDictionary->new_from_file($path);
    }
    return Compress::Zstd::DecompressionDictionary->new($self->{+COMPRESSION_DICTIONARY});
}

sub _compress {
    my ($self, $data) = @_;
    if ($self->_has_dict) {
        require Compress::Zstd::CompressionContext;
        my $ctx   = $self->{_compression_ctx}   //= Compress::Zstd::CompressionContext->new;
        my $cdict = $self->{_compression_cdict} //= $self->_build_cdict;
        return $ctx->compress_using_dict($data, $cdict);
    }
    return Compress::Zstd::compress($data, $self->{+COMPRESSION_LEVEL} // 3);
}

sub _decompress {
    my ($self, $data) = @_;
    my $out;
    if ($self->_has_dict) {
        require Compress::Zstd::DecompressionContext;
        my $ctx   = $self->{_decompression_ctx}   //= Compress::Zstd::DecompressionContext->new;
        my $ddict = $self->{_decompression_ddict} //= $self->_build_ddict;
        $out = $ctx->decompress_using_dict($data, $ddict);
    }
    else {
        $out = Compress::Zstd::decompress($data);
    }
    $self->throw_invalid("zstd decompression failed") unless defined $out;
    return $out;
}

sub eof {
    my $self = shift;

    $self->throw_invalid() if $self->{+INVALID_STATE};

    return 0 if $self->fill_buffer;
    return 0 unless $self->{+EOF};
    return 0 if $self->{+IN_BUFFER_SIZE};

    if (my $buffer = $self->{+MIXED_BUFFER}) {
        return 0 if defined($buffer->{lines}) && length($buffer->{lines});
        return 0 if defined($buffer->{burst}) && length($buffer->{burst});
    }

    return 1;
}

sub _fh_mode {
    my $self = shift;
    my ($fh) = @_;

    my $mode = fcntl($fh, Fcntl::F_GETFL(), 0) // return undef;
    return '<&' if $mode == Fcntl::O_RDONLY();
    return '>&' if $mode == Fcntl::O_WRONLY();
    return undef;
}

my %MODE_TO_DIR = (
    '<&'  => RH(),
    '<&=' => RH(),
    '>&'  => WH(),
    '>&=' => WH(),
);
sub _mode_to_dir {
    my $self = shift;
    my ($mode) = @_;
    return $MODE_TO_DIR{$mode};
}

sub _check_params {
    my ($class, %params) = @_;
    croak "IO::Select is not installed, cannot enable use_io_select"
        if $params{+USE_IO_SELECT} && !HAVE_IO_SELECT;

    if (defined(my $algo = $params{+COMPRESSION})) {
        croak "Unknown compression algorithm '$algo'" unless $algo eq 'zstd';
        croak "compression => 'zstd' requires Compress::Zstd"
            unless eval { require Compress::Zstd; 1 };
    }

    croak "compression_dictionary and compression_dictionary_file are mutually exclusive"
        if defined($params{+COMPRESSION_DICTIONARY}) && defined($params{+COMPRESSION_DICTIONARY_FILE});

    croak "compression_dictionary requires compression to be enabled"
        if (defined($params{+COMPRESSION_DICTIONARY}) || defined($params{+COMPRESSION_DICTIONARY_FILE}))
            && !defined($params{+COMPRESSION});
}

sub read_fifo {
    my $class = shift;
    my ($fifo, %params) = @_;

    $class->_check_params(%params);
    croak "File '$fifo' is not a pipe (-p check)" unless -p $fifo;

    open(my $fh, '+<', $fifo) or die "Could not open fifo ($fifo) for reading: $!";
    binmode($fh);

    return $class->_new_from_params(\%params, RH() => $fh);
}

sub write_fifo {
    my $class = shift;
    my ($fifo, %params) = @_;

    $class->_check_params(%params);
    croak "File '$fifo' is not a pipe (-p check)" unless -p $fifo;

    open(my $fh, '>', $fifo) or die "Could not open fifo ($fifo) for writing: $!";
    binmode($fh);

    return $class->_new_from_params(\%params, WH() => $fh);
}

sub from_fh {
    my $class = shift;

    # Mode is optional: from_fh($fh, %params) or from_fh($mode, $fh, %params).
    my $mode;
    $mode = shift if @_ && !ref($_[0]) && $MODE_TO_DIR{$_[0]};
    my $ifh = shift;
    my %params = @_;

    $class->_check_params(%params);

    croak "Filehandle is not a pipe (-p check)" unless -p $ifh;

    $mode //= $class->_fh_mode($ifh) // croak "Could not determine filehandle mode, please specify '>&' or '<&'";
    my $dir = $class->_mode_to_dir($mode) // croak "Invalid mode: $mode";

    open(my $fh, $mode, $ifh) or croak "Could not clone ($mode) filehandle: $!";
    binmode($fh);

    return $class->_new_from_params(\%params, $dir => $fh);
}

sub from_fd {
    my $class = shift;
    my ($mode, $fd, %params) = @_;

    $class->_check_params(%params);

    my $dir = $class->_mode_to_dir($mode) // croak "Invalid mode: $mode";
    open(my $fh, $mode, $fd) or croak "Could not open ($mode) fd$fd: $!";

    croak "Filehandle is not a pipe (-p check)" unless -p $fh;

    binmode($fh);
    return $class->_new_from_params(\%params, $dir => $fh);
}

sub _new_from_params {
    my ($class, $params, @handles) = @_;

    my $mixed = delete $params->{mixed_data_mode};

    my $self = bless({%$params, @handles}, $class);
    $self->set_mixed_data_mode() if $mixed;

    return $self;
}

sub new {
    my $class = shift;
    my (%params) = @_;

    $class->_check_params(%params);

    my ($rh, $wh);
    pipe($rh, $wh) or die "Could not create pipe: $!";

    binmode($wh);
    binmode($rh);

    return $class->_new_from_params(\%params, RH() => $rh, WH() => $wh);
}

sub pair {
    my $class = shift;
    my (%params) = @_;

    $class->_check_params(%params);

    my ($rh, $wh);
    pipe($rh, $wh) or die "Could not create pipe: $!";

    binmode($wh);
    binmode($rh);

    my $r = $class->_new_from_params({%params}, RH() => $rh);
    my $w = $class->_new_from_params({%params}, WH() => $wh);

    return ($r, $w);
}

sub set_mixed_data_mode {
    my $self = shift;

    $self->throw_invalid() if $self->{+INVALID_STATE};

    $self->read_blocking(0) if $self->{+RH};

    $self->{+BURST_PREFIX}  //= "\x0E";    # Shift out
    $self->{+BURST_POSTFIX} //= "\x0F";    # Shift in
    $self->{+MESSAGE_KEY}   //= "\x10";    # Data link escape
}

sub set_compression {
    my $self = shift;
    my ($algo, $level) = @_;

    if (!defined $algo) {
        delete $self->{+COMPRESSION};
        delete $self->{+COMPRESSION_LEVEL};
        delete $self->{_compression_ctx};
        delete $self->{_compression_cdict};
        delete $self->{_decompression_ctx};
        delete $self->{_decompression_ddict};
        delete $self->{_compress_cache};
        return;
    }

    croak "Unknown compression algorithm '$algo'" unless $algo eq 'zstd';
    croak "compression => 'zstd' requires Compress::Zstd"
        unless eval { require Compress::Zstd; 1 };

    $self->{+COMPRESSION}       = $algo;
    # Omitted $level preserves the previously set level; pass undef to
    # set_compression(undef) first if a full reset back to the default is desired.
    $self->{+COMPRESSION_LEVEL} = $level if defined $level;

    # Cached objects depend on level / dict; force rebuild.
    delete $self->{_compression_ctx};
    delete $self->{_compression_cdict};
    delete $self->{_decompression_ctx};
    delete $self->{_decompression_ddict};
    delete $self->{_compress_cache};

    return;
}

sub set_compression_dictionary {
    my ($self, $bytes) = @_;
    if (defined $bytes) {
        croak "compression_dictionary requires compression to be enabled"
            unless defined $self->{+COMPRESSION};
        $self->{+COMPRESSION_DICTIONARY} = $bytes;
        delete $self->{+COMPRESSION_DICTIONARY_FILE};
    }
    else {
        delete $self->{+COMPRESSION_DICTIONARY};
    }
    delete $self->{_compression_cdict};
    delete $self->{_decompression_ddict};
    delete $self->{_compress_cache};
    return;
}

sub set_compression_dictionary_file {
    my ($self, $path) = @_;
    if (defined $path) {
        croak "compression_dictionary requires compression to be enabled"
            unless defined $self->{+COMPRESSION};
        $self->{+COMPRESSION_DICTIONARY_FILE} = $path;
        delete $self->{+COMPRESSION_DICTIONARY};
    }
    else {
        delete $self->{+COMPRESSION_DICTIONARY_FILE};
    }
    delete $self->{_compression_cdict};
    delete $self->{_decompression_ddict};
    delete $self->{_compress_cache};
    return;
}

sub set_keep_compressed {
    my ($self, $val) = @_;
    $self->{+KEEP_COMPRESSED} = $val ? 1 : 0;
    return;
}

sub get_line_burst_or_data {
    my $self = shift;
    my %params = @_;

    my $rh = $self->{+RH} // croak "Not a read handle";

    my $prefix  = $self->{+BURST_PREFIX}  // croak "missing 'burst_prefix', not in mixed_data_mode";
    my $postfix = $self->{+BURST_POSTFIX} // croak "missing 'burst_postfix', not in mixed_data_mode";
    my $key     = $self->{+MESSAGE_KEY}   // croak "missing 'message_key', not in mixed_data_mode";

    my $buffer = $self->{+MIXED_BUFFER} //= {
        lines      => '',
        burst      => '',
        in_burst   => 0,
        in_message => 0,
        strip_term => 0,
    };

    my $peek;

    while (1) {
        $self->throw_invalid('Incomplete message received before EOF')
            if $self->eof && (keys(%{$self->{+STATE}->{buffers}}) || keys (%{$self->{+STATE}->{parts}}));

        if($buffer->{lines} || length($buffer->{lines})) {
            # Look for a complete line
            my ($line, $term);
            ($line, $term, $buffer->{lines}) = split /(\r?\n|\r\n?)/, $buffer->{lines}, 2;

            return (line => "${line}${term}") if $term;
            return (line => $line) if $self->{+EOF} && !$self->{+IN_BUFFER_SIZE} && defined($line) && length($line);

            $buffer->{lines} = $line;
            $peek = $line if $params{peek_line} && defined($line) && length($line);
        }

        if ($buffer->{in_message}) {
            my ($id, $message) = $self->_extract_message(one_part_only => 1);

            unless(defined $id) {
                next unless $self->{+EOF} && !$self->{+IN_BUFFER_SIZE};
                $self->throw_invalid('Incomplete burst data received before end of pipe');
            }

            $buffer->{strip_term}++;
            $buffer->{in_message} = 0;
            if (defined $message) {
                if ($self->{+COMPRESSION}) {
                    my $compressed   = $message;
                    my $decompressed = $self->_decompress($compressed);
                    return (message => $decompressed, compressed => $compressed)
                        if $self->{+KEEP_COMPRESSED};
                    return (message => $decompressed);
                }
                return (message => $message);
            }
        }

        if ($buffer->{strip_term}) {
            my $term = $self->_get_from_buffer(1, eof_invalid => 'EOF before message terminator') // return;

            $self->throw_invalid("No message terminator") unless $term eq $postfix;
            $buffer->{strip_term}--;
        }

        if ($buffer->{in_burst}) {
            my $peek = $self->_peek_from_buffer(1, eof_invalid => 'Incomplete burst data received before end of pipe') // next;

            if ($peek eq $key) {
                $self->_get_from_buffer(1); # Strip the key
                $buffer->{in_message} = 1;
                $buffer->{in_burst} = 0;
                next;
            }

            $buffer->{burst} //= '';
            my ($burst_data, $term);
            ($burst_data, $term, $self->{+IN_BUFFER}) = split /(\Q$postfix\E)/, $self->{+IN_BUFFER}, 2;
            $buffer->{burst} .= $burst_data;

            if ($term) {
                $self->{+IN_BUFFER_SIZE} = length($self->{+IN_BUFFER});
                $buffer->{in_burst} = 0;
                my $compressed = delete $buffer->{burst};
                if ($self->{+COMPRESSION}) {
                    my $decompressed = $self->_decompress($compressed);
                    return (burst => $decompressed, compressed => $compressed)
                        if $self->{+KEEP_COMPRESSED};
                    return (burst => $decompressed);
                }
                return (burst => $compressed);
            }
            else {
                $self->{+IN_BUFFER_SIZE} = 0;
            }

            $self->throw_invalid('Incomplete burst data received before end of pipe') if $self->{+EOF};
        }

        unless ($self->{+IN_BUFFER_SIZE} || $self->fill_buffer()) {
            return (peek => $peek) if $peek && !$self->{+EOF};

            return unless $self->{+EOF};

            # Do at least one more iteration after EOF
            return if $buffer->{+EOF}++;

            # But do not try to split the empty buffer
            next;
        }

        # Look for the start of a burst, anything before a burst is line data
        my $linedata;
        ($linedata, $buffer->{in_burst}, $self->{+IN_BUFFER}) = split /(\Q$prefix\E)/, $self->{+IN_BUFFER}, 2;
        $buffer->{lines} .= $linedata if defined $linedata;

        if ($buffer->{in_burst}) {
            $self->{+IN_BUFFER_SIZE} -= length($linedata) + length($buffer->{in_burst});
        }
        else {
            $self->{+IN_BUFFER_SIZE} = 0;
        }
    }
}

# This is a heavily modified version of a pattern suggested on stack-overflow
# and also used in Win32::PowerShell::IPC.
my $peek_named_pipe;
sub _win32_pipe_ready {
    my $self = shift;
    my $wh = Win32API::File::FdGetOsFHandle(fileno($self->{+RH}));

    my $buf = "";
    my $buflen = 0;

    $peek_named_pipe //= Win32::API->new("kernel32", 'PeekNamedPipe', 'NPIPPP', 'N')
        || die "Can't load PeekNamedPipe from kernel32.dll";

    my $got    = pack('L', 0);
    my $avail  = pack('L', 0);
    my $remain = pack('L', 0);

    my $ret = $peek_named_pipe->Call($wh, $buf, $buflen, $got, $avail, $remain);

    $self->{+EOF} = 1 if $ret == 0;

    return unpack('L', $avail);
}

my $set_named_pipe_handle_state;
sub _win32_set_pipe_state {
    my $self = shift;
    my ($state) = @_;
    my $wh = Win32API::File::FdGetOsFHandle(fileno($self->{+WH}));

    $set_named_pipe_handle_state //= Win32::API->new("kernel32", 'SetNamedPipeHandleState', 'NPPP', 'N')
        || die "Can't load SetNamedPipeHandleState from kernel32.dll";

    # Block or non-block?
    my $lpmode = $state ? pack('L', 0x00000000) : pack('L', 0x00000001);

    my $ret = $set_named_pipe_handle_state->Call($wh, $lpmode, +0, +0);

    return $ret;
}

sub read_blocking {
    my $self = shift;
    my $rh   = $self->{+RH} or croak "Not a reader";

    ($self->{+READ_BLOCKING}) = @_ if @_;

    unless (IS_WIN32) {
        $rh->blocking(@_);
    }

    return $self->{+READ_BLOCKING};
}

sub write_blocking {
    my $self = shift;
    my $wh   = $self->{+WH} or croak "Not a writer";

    return $self->{+WRITE_BLOCKING} unless @_;

    my ($val) = @_;
    $self->{+WRITE_BLOCKING} = $val;

    if (IS_WIN32) {
        $self->_win32_set_pipe_state(@_) if @_;
    }
    else {
        my $flags = fcntl($wh, &Fcntl::F_GETFL, 0);    # Get the current flags
        die $! unless defined $flags;
        if   ($val) { $flags &= ~&Fcntl::O_NONBLOCK }  # Clear O_NONBLOCK
        else        { $flags |= &Fcntl::O_NONBLOCK }   # Set O_NONBLOCK
        fcntl($wh, &Fcntl::F_SETFL, $flags) || die $!;
    }

    return $self->{+WRITE_BLOCKING};
}

sub blocking {
    my $self = shift;

    if ($self->{+RH} && !$self->{+WH}) {
        return $self->read_blocking(@_);
    }
    elsif ($self->{+WH} && !$self->{+RH}) {
        return $self->write_blocking(@_);
    }

    my $r = $self->read_blocking(@_);
    my $w = $self->write_blocking(@_);

    return 1 if $r && $w;
    return 0 if !$r && !$w;
    return undef;
}

sub size {
    my $self = shift;
    return unless defined &Fcntl::F_GETPIPE_SZ;
    my $fh = $self->{+WH} // $self->{+RH};
    fcntl($fh, Fcntl::F_GETPIPE_SZ(), 0);
}

sub resize {
    my $self = shift;
    my ($size) = @_;

    return unless defined &Fcntl::F_SETPIPE_SZ;
    my $fh = $self->{+WH} // $self->{+RH};

    # Force numeric: fcntl(F_SETPIPE_SZ, $string) is interpreted as
    # buffer-mode and silently fails with EINVAL.
    fcntl($fh, Fcntl::F_SETPIPE_SZ(), $size + 0);
}

my $ONE_MB = 1 * 1024 * 1024;

sub max_size {
    return $ONE_MB unless -e '/proc/sys/fs/pipe-max-size';

    open(my $max, '<', '/proc/sys/fs/pipe-max-size') or return $ONE_MB;
    chomp(my $val = <$max>);
    close($max);
    # Force numeric. <$max> returns a string; passing it to
    # fcntl(F_SETPIPE_SZ) directly triggers the same EINVAL bug
    # resize() guards against. Numify here so any caller of
    # max_size() that hands the result to fcntl gets an int.
    return ($val + 0) || $ONE_MB;
}

sub resize_or_max {
    my $self = shift;
    my ($size) = @_;
    $size = min($size, $self->max_size);
    $self->resize($size);
}

sub is_reader {
    my $self = shift;
    return 1 if $self->{+RH} && !$self->{+WH};
    return undef;
}

sub is_writer {
    my $self = shift;
    return 1 if $self->{+WH} && !$self->{+RH};
    return undef;
}

sub clone_writer {
    my $self = shift;
    my $class = blessed($self);
    open(my $fh, '>&:raw', $self->{+WH}) or die "Could not clone filehandle: $!";
    return bless({WH() => $fh}, $class);
}

sub clone_reader {
    my $self = shift;
    my $class = blessed($self);
    open(my $fh, '<&:raw', $self->{+RH}) or die "Could not clone filehandle: $!";
    return bless({RH() => $fh}, $class);
}

sub writer {
    my $self = shift;

    croak "pipe was set to reader, cannot set to writer" unless $self->{+WH};

    return 1 unless $self->{+RH};

    close(delete $self->{+RH});
    return 1;
}

sub reader {
    my $self = shift;

    croak "pipe was set to writer, cannot set to reader" unless $self->{+RH};

    return 1 unless $self->{+WH};

    $self->_flush_before_close;
    close(delete $self->{+WH});
    return 1;
}

# Buffered bursts from non-blocking writes must not be silently dropped when
# the write handle goes away (DESTROY would also croak trying to flush them
# without a handle).
sub _flush_before_close {
    my $self = shift;
    return if $self->{+HIT_EPIPE} || $self->{+INVALID_STATE};
    $self->flush(blocking => 1) if $self->pending_output;
}

sub close {
    my $self = shift;
    if ($self->{+WH}) {
        $self->_flush_before_close;
        close(delete $self->{+WH});
    }
    close(delete $self->{+RH}) if $self->{+RH};
    return;
}

my $psize = 16; # 32bit pid, 32bit tid, 32 bit size, 32 bit int part id;
my $dsize = PIPE_BUF - $psize;

sub delimiter_size {
    return $_[0]->{+DELIMITER_SIZE} if defined $_[0]->{+DELIMITER_SIZE};
    return $_[0]->{+DELIMITER_SIZE} //= bytes::length($_[0]->{+BURST_PREFIX} // '') + bytes::length($_[0]->{+BURST_POSTFIX} // '');
}

# The typical fits_in_burst() then write_burst() sequence would compress the
# same payload twice; remember the last result.
sub _compress_cached {
    my ($self, $data) = @_;

    my $cache = $self->{_compress_cache};
    return $cache->[1] if $cache && $cache->[0] eq $data;

    my $out = $self->_compress($data);
    $self->{_compress_cache} = [$data, $out];
    return $out;
}

sub fits_in_burst {
    my $self = shift;
    my ($data) = @_;

    $data = $self->_compress_cached($data) if $self->{+COMPRESSION};

    my $size = bytes::length($data) + ($self->{+DELIMITER_SIZE} // $self->delimiter_size);
    return undef unless $size <= PIPE_BUF;

    return $size;
}

sub write_burst {
    my $self = shift;
    my ($data) = @_;

    $data = $self->_compress_cached($data) if $self->{+COMPRESSION};

    my $size = bytes::length($data) + ($self->{+DELIMITER_SIZE} // $self->delimiter_size);
    return undef unless $size <= PIPE_BUF;

    push @{$self->{+OUT_BUFFER} //= []} => [$data, $size];
    $self->flush();

    return 1;
}

sub DESTROY {
    my $self = shift;
    local ($., $@, $!, $^E, $?);
    return if $self->{+HIT_EPIPE} || $self->{+INVALID_STATE};
    $self->flush(blocking => 1) if $self->{+WH} && $self->pending_output;
}

sub pending_output {
    my $self = shift;
    my $buffer = $self->{+OUT_BUFFER} or return 0;
    return 0 unless @$buffer;
    return 1;
}

sub flush {
    my $self = shift;
    my %params = @_;
    my $blocking = $params{blocking} // $self->{+WRITE_BLOCKING} // 1;

    my $buffer = $self->{+OUT_BUFFER} // return;

    while (@$buffer) {
        my $set = $buffer->[0];
        my $got = $self->_write_burst(@$set);

        return unless $blocking || defined $got;
        next unless defined $got;

        shift @$buffer;
    }

    return;
}

sub _write_burst {
    my $self = shift;
    my ($data, $size) = @_;

    my $wh = $self->{+WH} or croak "Cannot call write on a pipe reader";

    croak "Disconnected pipe" if $self->{+HIT_EPIPE};

    my $prefix  = $self->{+BURST_PREFIX}  // '';
    my $postfix = $self->{+BURST_POSTFIX} // '';

    $data = "${prefix}${data}${postfix}" if length($prefix) || length($postfix);

    my $wrote;
    SWRITE: {
        $wrote = syswrite($wh, $data, $size);

        # $! is only meaningful when syswrite fails.
        unless (defined $wrote) {
            if ($! == EPIPE || (IS_WIN32 && $! == 22)) {
                $self->{+HIT_EPIPE} = 1;
                delete $self->{+OUT_BUFFER};
                croak "Disconnected pipe";
            }
            return undef if $NONBLOCK_ERRNO{0 + $!} || (IS_WIN32 && $! == 28);    # NON-BLOCKING
            redo SWRITE if $RETRY_ERRNO{0 + $!};
            $self->throw_invalid("syswrite failed: $!");
        }

        redo SWRITE unless $wrote;
        last SWRITE if $wrote == $size;
        die "partial write: $wrote vs $size: $!";
    }

    return $wrote;
}

sub _adjusted_dsize {
    my $self = shift;

    return $self->{+ADJUSTED_DSIZE} if defined $self->{+ADJUSTED_DSIZE};

    my $message_key = $self->{+MESSAGE_KEY}   // '';
    my $prefix      = $self->{+BURST_PREFIX}  // '';
    my $postfix     = $self->{+BURST_POSTFIX} // '';

    my $fix_size = bytes::length($prefix) + bytes::length($postfix) + bytes::length($message_key);
    return $self->{+ADJUSTED_DSIZE} = $dsize - $fix_size;
}

sub write_message {
    my $self = shift;
    my ($data) = @_;

    $data = $self->_compress($data) if $self->{+COMPRESSION};

    my $tid            = _get_tid();
    my $message_key    = $self->{+MESSAGE_KEY}    // '';
    my $adjusted_dsize = $self->{+ADJUSTED_DSIZE} // $self->_adjusted_dsize;
    my $dtotal         = bytes::length($data);

    my $parts = int($dtotal / $adjusted_dsize);
    $parts++ if $dtotal % $adjusted_dsize;

    my $id = $parts - 1;

    # Unwinding the loop for a 1-part message for micro-optimization
    if ($parts == 1) {
        my $bytes = $data;
        my $size  = $dtotal;
        my $out   = $message_key . pack("l2L2", $$, $tid, $id--, $size) . $bytes;

        my $out_size = $dtotal + ($self->{+DELIMITER_SIZE} // $self->delimiter_size) + $psize + ($message_key ? 1 : 0);

        push @{$self->{+OUT_BUFFER} //= []} => [$out, $out_size];
    }
    else {
        for (my $part = 0; $part < $parts; $part++) {
            my $bytes = bytes::substr($data, $part * $adjusted_dsize, $adjusted_dsize);
            my $size  = bytes::length($bytes);

            my $out = $message_key . pack("l2L2", $$, $tid, $id--, $size) . $bytes;

            my $out_size = bytes::length($out) + ($self->{+DELIMITER_SIZE} // $self->delimiter_size);
            push @{$self->{+OUT_BUFFER} //= []} => [$out, $out_size];
        }
    }

    $self->flush();
    return $parts;
}

sub read_message {
    my $self = shift;
    my %params = @_;

    my ($id, $out) = $self->_extract_message(%params);

    return unless defined $id;

    return $out unless $self->{+COMPRESSION};

    if ($params{debug}) {
        my $compressed = $out->{message};
        $out->{message} = $self->_decompress($compressed);
        $out->{compressed} = $compressed if $self->{+KEEP_COMPRESSED};
        return $out;
    }

    my $compressed   = $out;
    my $decompressed = $self->_decompress($compressed);

    return ($decompressed, $compressed) if $self->{+KEEP_COMPRESSED} && wantarray;
    return $decompressed;
}

sub _extract_message {
    my $self   = shift;
    my %params = @_;

    my $state = $self->{+STATE} //= {};

    while (1) {
        unless ($state->{key}) {
            my $key_bytes = $self->_get_from_buffer($psize);
            unless (defined($key_bytes) && length($key_bytes)) {
                # Leftover bytes smaller than a header at EOF mean a writer
                # died mid-message; a plain return here would look like a
                # clean EOF and silently drop data.
                $self->throw_invalid("EOF inside message header (truncated message)")
                    if $self->{+EOF} && $self->{+IN_BUFFER_SIZE};
                return;
            }

            my %key;
            @key{qw/pid tid id size/} = unpack('l2L2', $key_bytes);
            $state->{key} = \%key;
        }

        my $key = $state->{key};

        my $data = $self->_get_from_buffer($key->{size}, eof_invalid => "EOF before end of message") // return;

        my $id   = $key->{id};
        my $tag  = join ':' => @{$key}{qw/pid tid/};
        push @{$state->{parts}->{$tag} //= []} => $id;
        $state->{buffers}->{$tag} = $state->{buffers}->{$tag} ? $state->{buffers}->{$tag} . $data : $data;

        delete $state->{key};

        unless ($id == 0) {
            return ($id, undef) if $params{one_part_only};
            next;
        }

        my $message = delete $state->{buffers}->{$tag};
        my $parts   = delete $state->{parts}->{$tag};

        return ($id, $message) unless $params{debug};

        return (
            $id,
            {
                message => $message,
                parts   => $parts,
                pid     => $key->{pid},
                tid     => $key->{tid},
            },
        );
    }
}

1;

__END__