eve.aio.posix.tcp

Async TCP listener and connection primitives for POSIX platforms.

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

Backpressure is modelled via an all-or-nothing send policy: send() either accepts all bytes (OK) or none (PRESSURE). A per-connection 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 send() call.
  • Pre-allocated ring buffer — set preallocatedSendBuffer for

    steady-state zero-allocation operation.

See Also

Backpressure Specification,

Buffer Ownership Redesign, Memory Management

alias AcceptCallback CloseCallback ConnectCallback ConnectionCallback DataCallback EofCallback ErrorCallback WritableCallback
struct AddrinfoRange TcpConnection TcpConnectionConfig TcpListener TcpListenerConfig
class TcpConnectionState TcpListenerState
fn accept4NonBlocking addrinfoRange copyStringz drainToEAGAIN queryLocalPort resolveErrno setNonBlocking socketError validated wouldBlock

Types 15

structTcpConnectionConfig

Connection-side buffer configuration.

Fields
size_t highWaterMarkMaximum queued bytes before `send` 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.
int recvBufferSize`SO_RCVBUF` value applied to accepted/connected sockets.
int sendBufferSize`SO_SNDBUF` value applied to accepted/connected sockets.
int windowClamp`TCP_WINDOW_CLAMP` value applied to accepted/connected sockets.
size_t readBufferSizePer-connection receive buffer size in bytes.
int keepAliveIdleSeconds before the first TCP keepalive probe is sent.
int keepAliveIntervalSeconds between subsequent TCP keepalive probes.
int keepAliveCountNumber of unacknowledged keepalive probes before the connection is considered dead.
bool keepAliveEnabledEnable TCP keepalive on the socket.
Methods
bool isValid() @property const pure @safe nothrow @nogcValidate the config.
structTcpListenerConfig

Listener-side socket configuration.

Fields
int backlogPending backlog size passed to `listen(2)`. -1 = OS default.
int deferAcceptSeconds`TCP_DEFER_ACCEPT` value applied to the listening socket.
bool reusePortEnable `SO_REUSEPORT` on the listening socket.
Methods
bool isValid() @property const pure @safe nothrow @nogcValidate the config.
private aliasConnectCallback = void delegate(ref TcpConnection connection) @safe
private aliasEofCallback = void delegate(ref TcpConnection connection) @safe
private aliasDataCallback = void delegate(ref TcpConnection connection, scope const(ubyte)[] data) @safe
private aliasWritableCallback = void delegate(ref TcpConnection connection) @safe
private aliasCloseCallback = void delegate(ref TcpConnection connection, SocketCloseReason reason) @safe
private aliasErrorCallback = void delegate(ref TcpConnection connection, int errorNumber) @safe
private aliasAcceptCallback = void delegate(ref TcpListener listener, Handle clientHandle) @safe
private aliasConnectionCallback = void delegate(ref TcpListener listener, ref TcpConnection connection) @safe
structTcpConnection

Async TCP connection wrapper.

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

Fields
private TcpConnectionState _state
Methods
TcpConnection create(TcpConnectionConfig config = TcpConnectionConfig.init) static @trusted nothrowCreate a detached connection wrapper.
void onConnect(ConnectCallback callback) @property @trustedSet the connect callback.
void onEof(EofCallback callback) @property @trustedSet the read half-close callback.
void onData(DataCallback callback) @property @trustedSet the inbound data callback.
void onWritable(WritableCallback callback) @property @trustedSet the writable callback.
void onClose(CloseCallback callback) @property @trustedSet the close callback.
void onError(ErrorCallback callback) @property @trustedSet the error callback.
bool isOpen() @property const @safe nothrow @nogcReport whether the wrapper currently owns an open socket.
SocketState state() @property const @safe nothrow @nogcReport the current connection state.
size_t sendQueueLen() @property const @safe nothrow @nogcReport the current queued send size.
bool isWritable() @property const @safe nothrow @nogcReport whether the connection is currently writable from the caller's perspective.
IpAddress remoteAddress() @property const @safe nothrow @nogcReport the remote address of the connection.
IpAddress localAddress() @property const @safe nothrow @nogcReport the local address of the connection.
AdoptStatus adopt(ref EventLoop loop, Handle clientHandle) @trusted nothrowAdopt an already-accepted client socket.
ConnectStatus connect(ref EventLoop loop, scope const(char)[] host, ushort port, CancelToken cancel = CancelToken .invalid) @trusted nothrowStart a non-blocking connection attempt.
SendStatus send(scope const(ubyte)[] data) @trustedSend bytes over the connection.
void pauseReading() @trusted nothrowPause socket read delivery.
void resumeReading() @trusted nothrowResume socket read delivery.
void close() @trusted nothrowClose the connection.
void shutdownWrite() @trusted nothrowShut the write direction down (graceful half-close, SHUT_WR).
void dispose() @trusted nothrowDispose the connection and its watcher registration.
bool track(Coord)(ref Coord coordinator) @trusted nothrowTrack this connection for graceful shutdown.
private TcpConnectionState mutableState() @trusted nothrow
structTcpListener

Async TCP listener wrapper.

Wraps a non-blocking TCP listening socket and integrates with the Layer 1 event loop for async accept operations.

Fields
private TcpListenerState _state
Methods
TcpListener create(TcpListenerConfig config = TcpListenerConfig.init) static @trustedCreate a detached listener wrapper.
void onAccept(AcceptCallback callback) @property @trustedSet the accept callback.
void onConnection(ConnectionCallback callback) @property @trustedSet the high-level connection callback.
bool isOpen() @property const @safe nothrow @nogcReport whether the listener currently owns an open socket.
ushort localPort() @property const @safe nothrow @nogcReport the effective local port once listening.
ListenStatus listen(ref EventLoop loop, scope const(char)[] host, ushort port) @trustedStart listening on the requested endpoint.
void close() @trusted nothrowClose the listener.
void dispose() @trusted nothrowDispose the listener and its watcher registration.
bool track(Coord)(ref Coord coordinator) @trusted nothrowTrack this listener for graceful shutdown.
private TcpListenerState mutableState() @trusted nothrow
private classTcpConnectionState : IShutdownTrackable
Fields
ConnectCallback onConnect
EofCallback onEof
DataCallback onData
WritableCallback onWritable
CloseCallback onClose
ErrorCallback onError
private EventLoop * _loop
private Handle _handle
private Token _token
private TcpConnectionConfig _config
private bool _readingPaused
private bool _watchingWrite
private bool _awaitingWritable
private bool _awaitingGlobalCapacity
private ulong _globalCapacitySubscription
private bool _closeDelivered
private bool _writeShutdown
private bool _readEof
private SendQueue _sendQueue
private SocketState _state
private IpAddress _remoteAddress
private IpAddress _localAddress
private CancelToken _cancelToken
private CancelToken _shutdownToken
private bool _draining
private ubyte[] _recvBuffer
Methods
void initiateShutdown(CancelToken token) @safe nothrow
void performDrain() @safe nothrow
void forceAbort() @safe nothrow
bool isClosed() @property const @safe nothrow @nogc
bool isOpen() @property const @safe nothrow @nogc
size_t sendQueueLen() @property const @safe nothrow @nogc
bool isWritable() @property const @safe nothrow @nogc
IpAddress remoteAddress() @property const @safe nothrow @nogc
IpAddress localAddress() @property const @safe nothrow @nogc
AdoptStatus adopt(ref EventLoop loop, Handle clientHandle) @trusted nothrow
ConnectStatus connect(ref EventLoop loop, scope const(char)[] host, ushort port, CancelToken cancel = CancelToken .invalid) @trusted nothrow
SendStatus send(scope const(ubyte)[] data) @trusted
private SendStatus backpressure() @trusted nothrow
private void globalCapacityAvailable() @safe nothrowRestores writability after global queue capacity is released.
private void cancelGlobalCapacityWait() @safe nothrowCancels a pending global queue-capacity notification.
void pauseReading() @trusted nothrow
void resumeReading() @trusted nothrow
void closeExplicitly() @trusted nothrow
void shutdownWrite() @trusted nothrowShut the write direction down (SHUT_WR half-close).
void completeWriteShutdown() @trusted nothrowDelivers the queued write half-close once output is flushed.
void invokeEof() @trusted nothrowInvokes the read half-close callback once.
void handleIo(ref EventLoop loop, Token token, IoReady ready) @safe nothrow
private TcpConnection owner() @safe nothrow
private void finishConnect() @trusted nothrow
private void readAvailable() @trusted nothrow
private void flushSendQueue() @trusted nothrow
private int refreshInterest() @trusted nothrow
private void fail(int errorNumber) @trusted nothrow
private void closeInternal(SocketCloseReason reason, int errorNumber, bool invokeCallback) @trusted nothrow
private void closeWithoutCallback() @trusted nothrow
private void invokeConnect() @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(SocketCloseReason reason) @trusted nothrow
private IpAddress toIpAddress(const sockaddr * addr) static @trusted nothrow @nogc
Constructors
this(TcpConnectionConfig config)
private classTcpListenerState : IShutdownTrackable
Fields
AcceptCallback onAccept
ConnectionCallback onConnection
private EventLoop * _loop
private Handle _handle
private Token _token
private TcpListenerConfig _config
private ushort _localPort
private CancelToken _shutdownToken
private Token _throttleToken
private bool _draining
private bool _throttled
Methods
void initiateShutdown(CancelToken token) @safe nothrow
void performDrain() @safe nothrow
void forceAbort() @safe nothrow
bool isClosed() @property const @safe nothrow @nogc
bool isOpen() @property const @safe nothrow @nogc
ListenStatus listen(ref EventLoop loop, scope const(char)[] host, ushort port) @trusted nothrow
void close() @trusted nothrow
void handleIo(ref EventLoop loop, Token token, IoReady ready) @safe nothrow
private TcpListener owner() @safe nothrow
private void acceptAvailable() @trusted nothrow
private void throttleAccept() @trusted nothrow
private void onThrottleExpired(ref EventLoop loop, Token token) @safe nothrow
private void invokeAccept(Handle clientHandle) @trusted nothrow
Constructors
this(TcpListenerConfig config)
private structAddrinfoRange

Range adapter for iterating over addrinfo linked list.

Fields
addrinfo * current
Methods
bool empty() @property const pure @safe nothrow @nogcCheck if the range is exhausted.
addrinfo * front() @property pure @safe nothrow @nogcGet the current addrinfo entry.
void popFront() pure @safe nothrow @nogcAdvance to the next entry.

Functions 11

private fnTcpConnectionConfig validated(TcpConnectionConfig config) pure @safe nothrow @nogcValidate a connection config, returning defaults if invalid.
private fnTcpListenerConfig validated(TcpListenerConfig config) pure @safe nothrow @nogcValidate a listener config, returning defaults if invalid.
private fnconst(char) * copyStringz(scope const(char)[] source, return char[] buffer) @trusted nothrowCopy a D string slice to a null-terminated C string buffer.
private fnint accept4NonBlocking(int srv, void * addr, uint * alen) @trusted nothrow @nogcAccept a new client socket with `SOCK_NONBLOCK | SOCK_CLOEXEC` atomically.
private fnint setNonBlocking(int fd) @trusted nothrowSet a file descriptor to non-blocking mode.
private fnbool wouldBlock(int value) pure @safe nothrow @nogcCheck if an errno value indicates the operation would block.
private fnptrdiff_t drainToEAGAIN(int fd) @trusted nothrow @nogcDrain a socket file descriptor to `EAGAIN` by reading and discarding.
private fnint socketError(int fd) @trusted nothrowQuery the socket error status.
private fnint resolveErrno(int status) @trusted nothrowConvert getaddrinfo error to errno equivalent.
private fnushort queryLocalPort(int fd) @trusted nothrowQuery the local port number for a bound socket.
private fnAddrinfoRange addrinfoRange(addrinfo * first) pure @safe nothrow @nogcCreate a range over an addrinfo linked list.