view release on metacpan or  search on metacpan

lib/Atomic/Pipe.pm  view on Meta::CPAN

=head1 SYNOPSIS

    use Atomic::Pipe;

    my ($r, $w) = Atomic::Pipe->pair;

    # Chunks will be set to the number of atomic chunks the message was split
    # into. It is fine to ignore the value returned, it will always be an
    # integer 1 or larger.
    my $chunks = $w->write_message("Hello");

    # $msg now contains "Hello";
    my $msg = $r->read_message;

    # Note, you can set the reader to be non-blocking:
    $r->blocking(0);

    # Writer too (but buffers unwritten items until your next write_burst(),
    # write_message(), or flush(), or will do a writing block when the pipe
    # instance is destroyed.
    $w->blocking(0);

    # $msg2 will be undef as no messages were sent, and blocking is turned off.
    my $msg2 = $r->read_message;

Fork example from tests:

    use Test2::V0;
    use Test2::Require::RealFork;
    use Test2::IPC;
    use Atomic::Pipe;

    my ($r, $w) = Atomic::Pipe->pair;

    # For simplicity
    $SIG{CHLD} = 'IGNORE';

    # Forks and runs your coderef, then exits.
    sub worker(&) { ... }

    worker { is($w->write_message("aa" x $w->PIPE_BUF), 3, "$$ Wrote 3 chunks") };
    worker { is($w->write_message("bb" x $w->PIPE_BUF), 3, "$$ Wrote 3 chunks") };
    worker { is($w->write_message("cc" x $w->PIPE_BUF), 3, "$$ Wrote 3 chunks") };

    my @messages = ();
    push @messages => $r->read_message for 1 .. 3;

    is(
        [sort @messages],
        [sort(('aa' x PIPE_BUF), ('bb' x PIPE_BUF), ('cc' x PIPE_BUF))],
        "Got all 3 long messages, not mangled or mixed, order not guaranteed"
    );

    done_testing;

Optional Zstd compression for bursts and messages (both ends must agree):

    my ($r, $w) = Atomic::Pipe->pair(compression => 'zstd');
    $w->write_message($big_payload);   # compressed on the wire
    my $msg = $r->read_message;        # decompressed transparently

See L</COMPRESSION> for details and options.

=head1 MIXED DATA MODE

Mixed data mode is a special use-case for Atomic::Pipe. In this mode the
assumption is that the writer end of the pipe uses the pipe as STDOUT or
STDERR, and as such a lot of random non-atomic prints can happen on the writer
end of the pipe. The special case is when you want to send atomic-chunks of
data inline with the random prints, and in the end extract the data from the
noise. The atomic nature of messages and bursts makes this possible.

Please note that mixed data mode makes use of 3 ASCII control characters:

=over 4

=item SHIFT OUT (^N or \x0E)

Used to start a burst

=item SHIFT IN (^O or \x0F)

Used to terminate a burst

=item DATA LINK ESCAPE (^P or \x10)

If this directly follows a SHIFT-OUT it marks the burst as being part of a
data-message.

=back

If the random prints include SHIFT OUT then they will confuse the read-side
parser and it will not be possible to extract data, in fact reading from the
pipe will become quite unpredictable. In practice this is unlikely to cause
issues, but printing a binary file or random noise could do it.

A burst may not include SHIFT IN as the SHIFT IN control+character marks the
end of a burst. A burst may also not start with the DATA LINK ESCAPE control
character as that is used to mark the start of a data-message.

data-messages may contain any data/characters/bytes as they messages include a
length so an embedded SHIFT IN will not terminate things early.

    # Create a pair in mixed-data mode
    my ($r, $w) = Atomic::Pipe->pair(mixed_data_mode => 1);

    # Open STDOUT to the write handle
    open(STDOUT, '>&', $w->{wh}) or die "Could not clone write handle: $!";

    # For sanity
    $wh->autoflush(1);

    print "A line!\n";

    print "Start a line ..."; # Note no "\n"

    # Any number of newlines is fine the message will send/receive as a whole.
    $w->write_burst("This is a burst message\n\n\n");

    # Data will be broken into atomic chunks and sent
    $w->write_message($data);

    print "... Finish the line we started earlier\n";

    my ($type, $data) = $r->get_line_burst_or_data;
    # Type: 'line'
    # Data: "A line!\n"

    ($type, $data) = $r->get_line_burst_or_data;
    # Type: 'burst'
    # Data: "This is a burst message\n\n\n"

    ($type, $data) = $r->get_line_burst_or_data;
    # Type: 'message'
    # Data: $data

    ($type, $data) = $r->get_line_burst_or_data;
    # Type: 'line'
    # Data: "Start a line ...... Finish the line we started earlier\n"

    # mixed-data mode is always non-blocking
    ($type, $data) = $r->get_line_burst_or_data;
    # Type: undef
    # Data: undef

You can also turn mixed-data mode after construction, but you must do so on both ends:

    $r->set_mixed_data_mode();
    $w->set_mixed_data_mode();

Doing so will make the pipe non-blocking, and will make all bursts/messages
include the necessary control characters.

=head1 COMPRESSION

C<Atomic::Pipe> can transparently compress B<bursts> and B<messages> (including
in mixed-data mode) with Zstandard. Plain C<print $wh ...> traffic is B<not>
compressed. Both ends of the pipe must be configured the same way; mismatch
produces protocol errors (or, in the case of mismatched dictionaries, silent
corruption -- see L</"Custom dictionary"> below).

Requires L<Compress::Zstd> (a soft / recommended dependency, loaded only when
compression is enabled).

=head2 Constructor options

All constructors (C<new>, C<pair>, C<from_fh>, C<from_fd>, C<read_fifo>,
C<write_fifo>) accept:

=over 4

=item compression => 'zstd'

Enable Zstd compression. Currently C<'zstd'> is the only supported algorithm;
any other value croaks at construction.

=item compression_level => $level

Zstd compression level, defaults to 3. Only meaningful when C<compression> is
enabled.

=item compression_dictionary => $bytes

Optional shared Zstd dictionary, supplied as raw bytes. Both ends must use the
same dictionary content. Mutually exclusive with C<compression_dictionary_file>.

=item compression_dictionary_file => $path

Same as C<compression_dictionary> but loaded from a file via
L<Compress::Zstd::CompressionDictionary/new_from_file>. The file is read on
demand.

=item keep_compressed => $bool

When set together with C<compression>, reads expose the on-wire compressed
bytes alongside the decompressed payload. See L</read_message> and
L</get_line_burst_or_data> for the exact return-shape changes. Has no effect
without C<compression>.

=back

=head2 Custom dictionary

Custom Zstd dictionaries can dramatically reduce frame size for small,
repetitive payloads. Either form (bytes or file) may be supplied at
construction or via L</set_compression_dictionary> /
L</set_compression_dictionary_file>.

B<Caveat:> raw zstd dictionaries do not embed a dict-ID. As a result a
B<mismatched> peer dictionary will silently decode to garbage rather than
fail. (Hard frame corruption -- truncated or invalid frames -- still raises
fatally.) Both ends must agree on byte-identical dictionary content.

=head2 Performance

Compression is not just a wire-size optimization for C<Atomic::Pipe>: when