eve.aio.posix.pipe

Async pipe primitives for POSIX platforms.

This module provides the POSIX implementation of anonymous pipe primitives using standard POSIX API. It works on Linux, BSD, and macOS.

Backpressure is modelled via an all-or-nothing write policy: write() either accepts all bytes (OK) or none (PRESSURE). A per-pipe send queue buffers data until the kernel can drain it. See the Backpressure Specification for details.

Two send-queue allocation modes are available:

  • Dynamic array (default) — may allocate per write() call.
  • Pre-allocated ring buffer — set preallocatedSendBuffer for

    steady-state zero-allocation operation.

See Also

Backpressure Specification,

Buffer Ownership Redesign, Memory Management

alias CloseCallback DataCallback ErrorCallback WritableCallback
struct Pipe PipeConfig PipePair
class PipeEndState
fn closePipe createPipe setCloseOnExec setNonBlocking validated wouldBlock

Types 8

structPipeConfig

Pipe buffer configuration.

Fields
size_t bufferSizeBuffer size for read operations.
size_t highWaterMarkMaximum queued bytes before `write` starts reporting pressure.
size_t lowWaterMarkQueue threshold below which `onWritable` may fire again.
size_t preallocatedSendBufferPre-allocate a fixed-capacity ring buffer for the send queue.
Methods
bool isValid() @property const pure @safe nothrow @nogcValidate the config.
private aliasDataCallback = void delegate(ref Pipe pipe, scope const(ubyte)[] data) @safe
private aliasWritableCallback = void delegate(ref Pipe pipe) @safe
private aliasCloseCallback = void delegate(ref Pipe pipe, PipeCloseReason reason) @safe
private aliasErrorCallback = void delegate(ref Pipe pipe, int errorNumber) @safe
structPipe

Async pipe endpoint wrapper.

Wraps a non-blocking pipe file descriptor and integrates with the Layer 1 event loop for async read/write operations with backpressure support.

A pipe has two ends: a read end and a write end. Use adoptRead to set up the read end, or adoptWrite to set up the write end.

Fields
PipeEndState _state
Methods
Pipe create(PipeConfig config = PipeConfig.init) static @trustedCreate a detached pipe endpoint wrapper.
void onData(DataCallback callback) @safeRegister callback for incoming data (read end only).
void onWritable(WritableCallback callback) @safeRegister callback when write buffer drains (write end only).
void onClose(CloseCallback callback) @safeRegister callback for close events.
void onError(ErrorCallback callback) @safeRegister callback for error conditions.
bool isOpen() @property const @safe nothrow @nogcCheck if the pipe endpoint is open.
PipeState state() @property const @safe nothrow @nogcGet the current pipe state.
int adoptRead(ref EventLoop loop, Handle readEnd) @trustedAdopt the read end of a pipe.
int adoptWrite(ref EventLoop loop, Handle writeEnd) @trustedAdopt the write end of a pipe.
SendResult write(scope const(ubyte)[] data) @trustedWrite bytes to the pipe (write end only).
void pauseReading() @trusted nothrowPause data delivery from read end.
void resumeReading() @trusted nothrowResume data delivery from read end.
void close() @trusted nothrowClose the pipe endpoint.
void dispose() @trusted nothrowDispose the pipe endpoint and its watcher registration.
PipeEndState mutableState() @trusted
structPipePair

Represents an anonymous pipe pair.

Contains handles for both the read and write ends of the pipe.

Fields
Handle readEndHandle for the read end of the pipe.
Handle writeEndHandle for the write end of the pipe.
Methods
bool isValid() @property const pure @safe nothrow @nogcCheck if the pipe pair is valid.
private classPipeEndState
Fields
DataCallback onData
WritableCallback onWritable
CloseCallback onClose
ErrorCallback onError
EventLoop * _loop
Handle _handle
Token _token
PipeConfig _config
bool _isReadEnd
bool _readingPaused
bool _watchingWrite
bool _awaitingWritable
bool _closeDelivered
SendQueue _sendQueue
PipeState _state
Methods
bool isOpen() @property const @safe nothrow @nogc
int adoptRead(ref EventLoop loop, Handle readEnd) @trusted nothrow
int adoptWrite(ref EventLoop loop, Handle writeEnd) @trusted nothrow
SendResult write(scope const(ubyte)[] data) @trusted
void pauseReading() @trusted nothrow
void resumeReading() @trusted nothrow
void closeExplicitly() @trusted nothrow
void handleIo(ref EventLoop loop, Token token, IoReady ready) @safe nothrow
Pipe owner() @safe nothrow
void readAvailable() @trusted nothrow
void flushSendQueue() @trusted nothrow
int refreshInterest() @trusted nothrow
void fail(int errorNumber) @trusted nothrow
void closeInternal(PipeCloseReason reason, int errorNumber, bool invokeCallback) @trusted nothrow
void closeWithoutCallback() @trusted nothrow
void invokeData(scope const(ubyte)[] data) @trusted nothrow
void invokeWritable() @trusted nothrow
void invokeError(int errorNumber) @trusted nothrow
void invokeClose(PipeCloseReason reason) @trusted nothrow
Constructors
this(PipeConfig config)

Functions 6

fnPipePair createPipe() @trusted nothrow @nogcCreate an anonymous pipe pair.
fnvoid closePipe(ref PipePair pair) @trusted nothrow @nogcClose both ends of a pipe pair.
private fnPipeConfig validated(PipeConfig config) pure @safe nothrow @nogc
private fnint setNonBlocking(int fd) @trusted nothrow @nogc
private fnint setCloseOnExec(int fd) @trusted nothrow @nogc
private fnbool wouldBlock(int value) @safe nothrow @nogc