NAME

Linux::Event::IO::Pipe - Asynchronous pipes and FIFOs

SYNOPSIS

use v5.36;
use Linux::Event::Loop;
use Linux::Event::IO::Pipe;

pipe(my $read_fh, my $write_fh)
    or die "pipe: $!";

my $loop = Linux::Event::Loop->new;

my $pipe = Linux::Event::IO::Pipe->new(
    loop    => $loop,
    read_fh => $read_fh,

    on_data => sub ($self, $bytes) {
        say "Received: $bytes";
        $self->close;
        $loop->stop;
    },
);

syswrite($write_fh, "hello") == 5
    or die "syswrite: $!";

$loop->run;

DESCRIPTION

Linux::Event::IO::Pipe provides asynchronous I/O for Linux pipes and FIFOs.

A Pipe may be:

read-only
write-only
read/write using one handle
read/write using separate input and output handles

The separate-handle form is particularly useful for things such as child process stdin and stdout, where reading and writing naturally happen through different pipe descriptors.

A Pipe uses the same ordered-byte behavior as Linux::Event::IO::Sock::Stream.

That means it supports:

raw on_data callbacks
Linux::Event::Framer message framing
queued asynchronous output
backpressure
read pausing
timeouts and deadlines

TLS does not apply to Pipe objects.

CREATING A READ-ONLY PIPE

Supply read_fh:

my $pipe = Linux::Event::IO::Pipe->new(
    loop    => $loop,
    read_fh => $read_fh,

    on_data => sub ($self, $bytes) {
        print $bytes;
    },
);

The supplied handle must refer to a Linux pipe or FIFO.

The Pipe can receive input but cannot write because no output handle was supplied.

CREATING A WRITE-ONLY PIPE

Supply write_fh:

my $pipe = Linux::Event::IO::Pipe->new(
    loop     => $loop,
    write_fh => $write_fh,
);

Then write asynchronously with:

$pipe->write("hello");

No input callback is required because this Pipe has no readable direction.

USING SEPARATE READ AND WRITE HANDLES

A Pipe may use different handles for input and output:

my $pipe = Linux::Event::IO::Pipe->new(
    loop     => $loop,
    read_fh  => $stdout_from_child,
    write_fh => $stdin_to_child,

    on_data => sub ($self, $bytes) {
        ...
    },
);

This is useful when one logical communication channel is built from two one-way pipes.

For example, a child process may have:

parent writes -> child stdin

and:

child stdout -> parent reads

Linux::Event can treat those two descriptors as one Pipe object.

USING ONE HANDLE FOR BOTH DIRECTIONS

When one handle supplies both readable and writable directions, use fh:

my $pipe = Linux::Event::IO::Pipe->new(
    loop => $loop,
    fh   => $fh,

    on_data => sub ($self, $bytes) {
        ...
    },
);

fh cannot be combined with read_fh or write_fh.

The supplied handle must be a Linux pipe or FIFO.

ATTACHING TO A LOOP

A Pipe can be attached during construction:

my $pipe = Linux::Event::IO::Pipe->new(
    loop    => $loop,
    read_fh => $read_fh,
    ...
);

or created detached:

my $pipe = Linux::Event::IO::Pipe->new(
    read_fh => $read_fh,
    ...
);

and added later:

$loop->add($pipe);

Both forms are normal Linux::Event APIs.

RECEIVING DATA

on_data

An unframed readable Pipe receives bytes through on_data:

on_data => sub ($self, $bytes) {
    print $bytes;
}

$bytes contains the next available part of the ordered byte stream.

A Pipe does not inherently know where your application's messages begin or end.

If the data has message boundaries, use Linux::Event::Framer.

MESSAGE FRAMING

A Pipe can use the same framing system as a Stream.

For example, a newline-delimited Pipe can be written as:

package LinePipe;

use parent 'Linux::Event::IO::Pipe';
use Linux::Event::Framer 'Delimiter', "\n";

sub on_message ($self, $message) {
    say "Received line: $message";
}

Then:

my $pipe = LinePipe->new(
    loop    => $loop,
    read_fh => $read_fh,
);

Incoming bytes are accumulated until a complete framed message is available.

The application receives on_message rather than having to maintain its own partial-input buffer.

Constructor callback with framing

Framing belongs to the subclass, but the message callback may still be supplied for one object:

my $prefix = 'received';

my $pipe = LinePipe->new(
    loop    => $loop,
    read_fh => $read_fh,

    on_message => sub ($self, $message) {
        say "$prefix: $message";
    },
);

The constructor callback overrides a same-named subclass method for that Pipe.

SENDING DATA

write($bytes)

$pipe->write("hello");

Send raw ordered bytes.

Linux::Event first attempts to write immediately.

If the output descriptor cannot accept all of the data, the remaining bytes are queued and written later.

Data remains in order.

send($payload)

For a framed Pipe:

$pipe->send($payload);

send applies the Pipe subclass's Linux::Event::Framer before queuing the resulting bytes.

For an unframed Pipe, use write.

BACKPRESSURE

A writable Pipe has an output queue.

When queued output grows past the high watermark, write or send begins returning false.

The supplied data is still accepted unless a configured hard queue limit would be exceeded.

When queued output later falls to the low watermark, on_drain is called:

on_drain => sub ($self) {
    # Producing more output is safe again.
}

This allows a producer to slow itself down instead of allowing the output queue to grow without bound.

EOF

on_eof

on_eof => sub ($self) {
    say "Input reached EOF";
}

Called when the readable side reaches end-of-file.

For a pipe, this commonly means that all writers for the other end have been closed.

EOF affects the readable direction and does not necessarily mean that the Pipe's writable direction has also ended.

PAUSING INPUT

pause_read

$pipe->pause_read;

Temporarily stop delivering input.

resume_read

$pipe->resume_read;

Resume input delivery.

Read timeout accounting is suspended while input is deliberately paused.

CALLBACKS

Pipe callbacks may be supplied directly to new or implemented as subclass methods.

The normal callbacks are:

on_data($pipe, $bytes): Raw unframed input arrived.
on_message($pipe, $message): One complete framed message arrived.
on_messages($pipe, $messages): A batch of framed messages arrived when message batching is enabled.
on_drain($pipe): Queued output fell to the low watermark after backpressure.
on_eof($pipe): The readable direction reached EOF.
on_error($pipe, $error): An asynchronous I/O or policy error occurred.
on_close($pipe): The Pipe closed.

A constructor callback overrides a same-named subclass method for that object.

APPLICATION DATA

An arbitrary value may be stored with the Pipe:

my $pipe = Linux::Event::IO::Pipe->new(
    read_fh => $read_fh,
    data    => $state,
    ...
);

This is useful for associating application state with the Pipe.

CLOSING DIRECTIONS

Because a Pipe may have separate read and write handles, its two directions can be controlled independently.

close_read

$pipe->close_read;

Close the readable direction immediately.

The writable direction may remain active.

close_write

$pipe->close_write;

Close the writable direction immediately.

The readable direction may remain active.

end

$pipe->end;

Allow already accepted output to drain, then end the writable direction.

Use end when queued data should be delivered before output is closed.

close

$pipe->close;

Close the entire Pipe immediately.

This is terminal for both directions.

DETACHING PIPE HANDLES

detach

my $handles = $pipe->detach;

Detach the Pipe from Linux::Event and return its still-open handles.

The return value is a hash reference containing:

read_fh
write_fh

as applicable.

For example:

my $handles = $pipe->detach;

my $read_fh  = $handles->{read_fh};
my $write_fh = $handles->{write_fh};

Detachment requires the output queue to be empty.

It is a terminal ownership transfer and does not invoke on_close.

SUBCLASSING

Subclassing is optional.

Constructor callbacks are usually simplest for one Pipe:

my $pipe = Linux::Event::IO::Pipe->new(
    read_fh => $read_fh,

    on_data => sub ($self, $bytes) {
        ...
    },
);

A subclass is useful when many Pipe objects share framing, callbacks, or tuning:

package ProtocolPipe;

use parent 'Linux::Event::IO::Pipe';
use Linux::Event::Framer 'Delimiter', "\n";

sub on_message ($self, $message) {
    ...
}

STREAM TUNING

Pipe uses the same ordered-byte tuning model as Stream.

Most applications should leave the defaults unchanged.

Reusable defaults may be declared in a Pipe subclass with stream_tuning:

package BulkPipe;

use parent 'Linux::Event::IO::Pipe';

sub stream_tuning ($class) {
    return (
        read_size         => 131_072,
        read_budget_bytes => 524_288,
        max_buffer        => 16_777_216,
    );
}

stream_tuning is subclass policy.

These values are not placed inside a constructor tuning hash.

Timeout and deadline values that Linux::Event allows as per-object overrides may be supplied directly to new.

read_size

Default: 65,536 bytes.

Maximum number of bytes requested by one native read.

read_budget_bytes

Default: 65,536 bytes.

Maximum amount of input read during one readiness turn before yielding to other Loop resources.

This is a fairness control.

A value of zero explicitly requests unlimited reading until the descriptor would block.

read_batch_bytes

Default: 0.

For an unframed Pipe, successful reads may be combined before on_data is called.

Zero preserves normal read callback boundaries.

This option cannot be used with framing.

message_batch_size

Default: 0.

For a framed Pipe, deliver up to this many complete messages together through on_messages.

Zero uses normal on_message delivery.

A positive value requires framing and an on_messages callback.

max_buffer

Default: 8,388,608 bytes.

Hard limit for retained input, incomplete framed data, and data retained for one message batch.

high_watermark

Default: 1,048,576 bytes.

Queued-output level at which write and send begin returning false to signal backpressure.

low_watermark

Default: 262,144 bytes.

After backpressure has occurred, on_drain fires when queued output falls to or below this level.

The low watermark cannot exceed the high watermark.

max_pending_bytes

Default: 0.

Hard limit on pending output.

Zero means no hard output-queue limit.

idle_timeout

Default: 0.

Maximum number of seconds without successful input or output progress.

Zero disables the timeout.

read_timeout

Default: 0.

Maximum number of seconds without inbound progress while reading is active.

A deliberate pause_read suspends this timeout.

Zero disables it.

write_timeout

Default: 0.

Maximum number of seconds without output progress while data remains queued.

Zero disables it.

PER-OBJECT TIMEOUTS AND DEADLINES

Timeout and deadline overrides that apply to one Pipe are supplied directly to new.

For example:

my $pipe = Linux::Event::IO::Pipe->new(
    loop    => $loop,
    read_fh => $read_fh,

    idle_timeout => 60,
    read_timeout => 10,
    deadline => {
        after     => 120,
        operation => 'read',
    },

    on_data => sub ($self, $bytes) {
        ...
    },
);

These options apply to this Pipe object.

They are not placed inside a nested tuning hash.

The supported deadline behavior is described in docs/ORDERED-BYTE-DEADLINES.md.

PERFORMANCE MODEL

Pipe uses the same native ordered-byte engine as Stream and TTY.

Callbacks and reusable subclass policy are resolved when the Pipe is created, rather than rediscovered for every read or write event.

Framing and buffering are handled before application callbacks are invoked.

These details normally require no application action.

	Global
`s`	Focus search bar
`?`	Bring up this help dialog

	GitHub
`g` `p`	Go to pull requests
`g` `i`	Go to GitHub issues (only if GitHub is preferred repository)

	POD
`g` `a`	Go to author
`g` `c`	Go to changes
`g` `i`	Go to issues
`g` `d`	Go to dist
`g` `r`	Go to repository/SCM
`g` `s`	Go to source
`g` `b`	Go to file browse

Search terms
module: (e.g. module:Plugin)
distribution: (e.g. distribution:Dancer auth)
author: (e.g. author:SONGMU Redis)
version: (e.g. version:1.00)

NAME

SYNOPSIS

DESCRIPTION

CREATING A READ-ONLY PIPE

CREATING A WRITE-ONLY PIPE

USING SEPARATE READ AND WRITE HANDLES

USING ONE HANDLE FOR BOTH DIRECTIONS

ATTACHING TO A LOOP

RECEIVING DATA

on_data

MESSAGE FRAMING

Constructor callback with framing

SENDING DATA

write($bytes)

send($payload)

BACKPRESSURE

EOF

on_eof

PAUSING INPUT

pause_read

resume_read

CALLBACKS

APPLICATION DATA

CLOSING DIRECTIONS

close_read

close_write

end

close

DETACHING PIPE HANDLES

detach

SUBCLASSING

STREAM TUNING

read_size

read_budget_bytes

read_batch_bytes

message_batch_size

max_buffer

high_watermark

low_watermark

max_pending_bytes

idle_timeout

read_timeout

write_timeout

PER-OBJECT TIMEOUTS AND DEADLINES

PERFORMANCE MODEL

SEE ALSO

Module Install Instructions

Keyboard Shortcuts