/ 
Atomic-Pipe-0.033
⭐ Starred 0 / README.md

NAME

Atomic::Pipe - Send atomic messages from multiple writers across a POSIX pipe.

DESCRIPTION

Normally if you write to a pipe from multiple processes/threads, the messages will come mixed together unpredictably. Some messages may be interrupted by parts of messages from other writers. This module takes advantage of some POSIX specifications to allow multiple writers to send arbitrary data down a pipe in atomic chunks to avoid the issue.

NOTE: This only works for POSIX compliant pipes on POSIX compliant systems. Also some features may not be available on older systems, or some platforms.

Also: https://man7.org/linux/man-pages/man7/pipe.7.html

POSIX.1 says that write(2)s of less than PIPE_BUF bytes must be
atomic: the output data is written to the pipe as a contiguous
sequence.  Writes of more than PIPE_BUF bytes may be nonatomic: the
kernel may interleave the data with data written by other processes.
POSIX.1 requires PIPE_BUF to be at least 512 bytes.  (On Linux,
PIPE_BUF is 4096 bytes.) [...]

Under the hood this module will split your message into small sections of slightly smaller than the PIPE_BUF limit. Each section is sent as 1 atomic chunk with a 16 byte header consisting of four 32-bit fields: the process id it came from, the thread id it came from, a chunk ID (in descending order, so if there are 3 chunks the first will have id 2, the second 1, and the final chunk is always 0 allowing a flush as it knows it is done) and the length of the data section to follow.

NOTE: Payloads are byte strings. If you have a wide-character (unicode) string, encode it (e.g. with "encode" in Encode) before passing it to write_message() or write_burst(); decode on the read side.

On the receiving end this module will read chunks and re-assemble them based on the header data. So the reader will always get complete messages. Note that message order is not guaranteed when messages are sent from multiple processes or threads. Though all messages from any given thread/process should be in order.

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, only for messages too big for one atomic write (the recommended way to use compression):

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

# Fits in one atomic part: sent as-is, no compression cost.
$w->write_message($small_payload);

# Larger than one atomic part (4080 bytes on Linux by default): compressed
# first, then split only if it is still too big.
$w->write_message($big_payload);

my $msg = $r->read_message;    # decompressed transparently when needed

Pass compression_threshold => $bytes to use a different cut-off.

Or compress every burst and message (both ends must agree):

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

See "COMPRESSION" for details and options.

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:

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.

COMPRESSION

Atomic::Pipe can transparently compress bursts and messages (including in mixed-data mode) with Zstandard. Plain print $wh ... traffic is 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 "Custom dictionary" below).

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

Constructor options

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

Compressing only large messages

With plain compression every burst and message is compressed. For small messages that is pure CPU cost: they already fit in one atomic write, and decompressing on a single reader is slower than reading them as-is. compress_large_only compresses a message only when it is longer than "compression_threshold", and only keeps the compressed form when it is actually smaller (already-compressed or random payloads go out plain).

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

That is the whole setup. It turns on zstd and uses the default threshold:

To use a different cut-off:

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

Each compressed message is marked by the high bit of the part-size field in its headers, so the reader knows which messages to decompress. Consequences:

Plain compression without compress_large_only keeps the original wire format, so it stays compatible with older versions.

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 set_compression_dictionary / set_compression_dictionary_file.

Caveat: raw zstd dictionaries do not embed a dict-ID. As a result a 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.

Performance

Compression trades CPU for wire size. Whether it is also faster depends on where the bottleneck is:

Versions before 0.033 had a reader bug that made reading large streams quadratic, especially in mixed-data mode. Compression shrank the stream and so hid most of it, which is why older versions of this documentation reported compression as up to ~1000x faster. That advice no longer applies.

The kernel pipe buffer size (see resize) has little effect on throughput.

Benchmark: streaming JSON objects

From bench/zstd_compression.pl in the distribution: one writer, one reader, mixed-data mode, 100 MB of synthetic JSON objects of 187 .. 10000 bytes (average ~5.1 KB, so most are larger than PIPE_BUF), 1 MB pipe:

mode        raw MB/s   wire MB    ratio
plain         530.70    100.00       -
L-3           276.39     35.61    2.81x
L1            240.18     30.56    3.27x
L3 (def)      208.35     30.25    3.31x
large-only    244.90     40.67    2.46x

("large-only" is compress_large_only at the default threshold and level.)

A separate run with 1 and 8 writers into one reader, MB/s. Large messages are ~1.6 MB:

large messages        plain    zstd   large-only
1 writer, mixed         883     482          503
1 writer, read_msg     1337     530          520
8 writers, mixed        878    1143         1112
8 writers, read_msg    1339    1148         1171

Small messages are ~1.2 KB, so large-only sends them all plain:

small messages        plain    zstd   large-only
1 writer, mixed         359     121          335
1 writer, read_msg      555     129          477
8 writers, mixed        344     119          334
8 writers, read_msg     649     140          587

Practical guidance

These results depend heavily on payload entropy, CPU and topology. Re-run bench/zstd_compression.pl against a representative slice of your own data before deciding.

METHODS

CLASS METHODS

OBJECT METHODS

PRIMARY INTERFACE

RESIZING THE PIPE BUFFER

On some newer linux systems it is possible to get/set the pipe size. On supported systems these allow you to do that, on other systems they are no-ops, and any that return a value will return undef.

Note: This has nothing to do with the similarly named PIPE_BUF which cannot be changed. This simply effects how much data can sit in a pipe before the writers block, it does not effect the max size of atomic writes.

SPLITTING THE PIPE INTO READER AND WRITER

If you used Atomic::Pipe->new() you need to now split the one object into readers and writers. These help you do that.

MIXED DATA MODE METHODS

COMPRESSION METHODS

SOURCE

The source code repository for Atomic-Pipe can be found at http://github.com/exodist/Atomic-Pipe.

MAINTAINERS

AUTHORS

COPYRIGHT

Copyright 2020 Chad Granum exodist7@gmail.com.

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

See http://dev.perl.org/licenses/