eve.aio.windows.pipe

Async pipe primitives for Windows.

This module provides the Windows implementation of anonymous pipe primitives. On Windows, anonymous pipes created with CreatePipe do not support overlapped I/O directly, so this implementation uses named pipes with unique names for event loop integration.

Backpressure is modelled via an all-or-nothing write policy: write() either accepts all bytes (OK) or rejects with PRESSURE. A per-pipe send queue buffers data until the kernel can drain it. When the queue drains below the low water mark, onWritable fires.

See Also

Backpressure Specification,

Buffer Ownership Redesign

var PIPE_READ_BUFFER_SIZE
alias CloseCallback DataCallback ErrorCallback WritableCallback
struct Pipe PipeConfig PipePair
class PipeEndState
fn closePipe createPipe formatPipeName formatUint mapWinError validated wouldBlock

Types 8

structPipeConfig

Pipe buffer configuration.

Fields
size_t bufferSizeSize of the internal read/write buffer.
size_t highWaterMarkMaximum send queue size before backpressure is applied.
size_t lowWaterMarkSend queue size at which backpressure is lifted and `onWritable` fires.
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 end wrapper.

Represents one end of a pipe - either the read or write end. Wraps a non-blocking pipe handle and integrates with the Layer 1 event loop for async read/write operations.

Fields
private PipeEndState _state
Methods
Pipe create(PipeConfig config = PipeConfig.init) static @trustedCreate a detached pipe wrapper.
void onData(DataCallback callback) @safe Register a callback for data available (read end). Fires when data arrives. The `data` parameter is borrowed — valid only for the duration of the callback. Copy it if needed beyond. Thread a...
void onWritable(WritableCallback callback) @safe Register a callback for writability (write end). Fires when the send queue drains below the low water mark after backpressure. Not called on initial open. Thread affinity: event loop thread on...
void onClose(CloseCallback callback) @safe Register a callback for pipe close. Fires when the peer closes the pipe or a fatal error occurs. Does not fire after `dispose()`. Thread affinity: event loop thread only. Non-reentrant. Does ...
void onError(ErrorCallback callback) @safeRegister a callback for pipe errors.
bool isOpen() @property const @safe nothrow @nogcCheck whether the pipe is open.
PipeState state() @property const @safe nothrow @nogcGet the current pipe state.
int adoptRead(ref EventLoop loop, Handle readEnd) @trustedAdopt a pipe handle for reading.
int adoptWrite(ref EventLoop loop, Handle writeEnd) @trustedAdopt a pipe handle for writing.
SendResult write(scope const(ubyte)[] data) @trustedWrite bytes to the pipe (write end only).
void pauseReading() @trusted nothrowPause read delivery.
void resumeReading() @trusted nothrowResume read delivery.
void close() @trusted nothrowClose the pipe end.
void dispose() @trusted nothrowDispose the pipe and its watcher registration.
private PipeEndState mutableState() @trusted
structPipePair

Represents a pair of pipe ends.

Fields
Handle readEndHandle for reading from the pipe.
Handle writeEndHandle for writing to the pipe.
Methods
bool isValid() @property const @safe nothrow @nogcCheck whether both pipe ends are valid.
private classPipeEndState

Internal state for a pipe end using IOCP overlapped I/O.

This implementation uses Windows overlapped I/O with completion ports instead of polling. Each pipe end has its own OVERLAPPED structure and buffers that remain valid until the I/O completes.

The pipe integrates with the event loop via a wakeup watcher that is notified when IOCP completions arrive.

Fields
DataCallback onData
WritableCallback onWritable
CloseCallback onClose
ErrorCallback onError
private EventLoop * _loopReference to the event loop this pipe is registered with.
private Handle _handleThe underlying Windows pipe handle.
private PipeConfig _configPipe configuration (buffer sizes, etc.).
private bool _isReadEndWhether this is the read end (`true`) or write end (`false`).
private bool _readingPausedWhether reading is paused (read end only).
private bool _closeDeliveredWhether the close callback has been delivered.
private PipeState _pipeStateCurrent lifecycle state of the pipe.
private OVERLAPPED _readOverlappedOverlapped structure for pending read operations.
private OVERLAPPED _writeOverlappedOverlapped structure for pending write operations.
private ubyte[PIPE_READ_BUFFER_SIZE] _readBufferRead buffer — must remain valid during overlapped read.
private ubyte[PIPE_READ_BUFFER_SIZE] _writeBufferWrite buffer — must remain valid during overlapped write.
private bool _readInProgressWhether a read operation is currently in progress.
private bool _writeInProgressWhether a write operation is currently in progress.
private bool _awaitingWritableWhether we're waiting for backpressure to clear before notifying writable.
private size_t _pendingWriteLenNumber of bytes in the current pending write.
private ubyte[] _sendQueueQueue of data waiting to be written.
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 nothrow
void pauseReading() @trusted nothrow
void resumeReading() @trusted nothrow
void closeExplicitly() @trusted nothrowExplicitly close this pipe end.
private void pipeCompletionCallback(void * context, OVERLAPPED * overlappedPtr, size_t bytesTransferred, DWORD error) static @trusted nothrowStatic IOCP completion callback.
private Pipe owner() @safe nothrow
private int associateWithIocp() @trusted nothrowAssociate the pipe handle with the event loop's IOCP.
private void submitRead() @trusted nothrowSubmit an overlapped read operation.
private void processReadCompletionData(DWORD bytesRead, DWORD error) @trusted nothrowProcess a completed read operation using IOCP completion data.
private void submitWrite() @trusted nothrowSubmit an overlapped write operation.
private void processWriteCompletionData(DWORD bytesWritten, DWORD error) @trusted nothrowProcess a completed write operation using IOCP completion data.
private void fail(int errorNumber) @trusted nothrow Transition the pipe to error state. Invokes the error callback and closes the pipe. Params: errorNumber = The error code to report.
private void closeInternal(PipeCloseReason reason, int errorNumber, bool invokeCallback) @trusted nothrowInternal close implementation.
private void closeWithoutCallback() @trusted nothrow
private void invokeData(scope const(ubyte)[] data) @trusted nothrow
private void invokeWritable() @trusted nothrow
private void invokeError(int errorNumber) @trusted nothrow
private void invokeClose(PipeCloseReason reason) @trusted nothrow
Constructors
this(PipeConfig config)

Functions 8

fnPipePair createPipe() @trusted nothrowCreate an anonymous pipe pair with overlapped I/O support.
fnvoid closePipe(ref PipePair pair) @trusted nothrowClose both ends of a pipe pair.
private fnPipeConfig validated(PipeConfig config) pure @safe nothrow @nogc
private fnconst(wchar) * formatPipeName(wchar[] buffer, uint counter) @trusted nothrow
private fnsize_t formatUint(wchar[] buffer, uint value) @trusted nothrow
private fnbool wouldBlock() @trusted nothrow
private fnbool wouldBlock(DWORD err) @trusted nothrow
private fnint mapWinError(DWORD winError) @trusted nothrow

Variables 1

private enumvarPIPE_READ_BUFFER_SIZE = 4096

Size of the internal read/write buffer for overlapped I/O.