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 ErrorCallback WritableCallback
struct AddrinfoRange TcpConnection TcpConnectionConfig TcpListener TcpListenerConfig
class TcpConnectionState TcpListenerState
fn accept4NonBlocking addrinfoRange copyStringz drainToEAGAIN mapAdoptError mapConnectError mapListenError queryLocalPort resolveErrno setNonBlocking socketError validated wouldBlock

Types 14

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 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 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.
AdoptResult adopt(ref EventLoop loop, Handle clientHandle) @trusted nothrowAdopt an already-accepted client socket.
ConnectResult connect(ref EventLoop loop, scope const(char)[] host, ushort port, CancelToken cancel = CancelToken .invalid) @trusted nothrowStart a non-blocking connection attempt.
SendResult 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 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.
ListenResult 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
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 _closeDelivered
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
AdoptResult adopt(ref EventLoop loop, Handle clientHandle) @trusted nothrow
ConnectResult connect(ref EventLoop loop, scope const(char)[] host, ushort port, CancelToken cancel = CancelToken .invalid) @trusted nothrow
SendResult send(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
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
ListenResult 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 14

private fnListenResult mapListenError(int err) pure @safe nothrow @nogc
private fnConnectResult mapConnectError(int err) pure @safe nothrow @nogc
private fnAdoptResult mapAdoptError(int err) pure @safe nothrow @nogc
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.