eve.aio.windows.tcp

Async TCP listener and connection primitives for Windows.

This module provides the Windows implementation of TCP networking primitives using Winsock2 API with IOCP overlapped I/O support.

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

Two I/O modes are available:

  • WSAPoll mode (default) — synchronous non-blocking I/O with edge

    notifications via the event loop poller.

  • IOCP_OVERLAPPED mode — true asynchronous I/O using AcceptEx,

    ConnectEx, WSARecv/WSASend with IOCP completion polling.

See Also

Backpressure Specification,

Buffer Ownership Redesign

var ACCEPT_ADDR_LEN ACCEPT_BUFFER_SIZE SOMAXCONN TCP_READ_BUFFER_SIZE
alias AcceptCallback CloseCallback ConnectCallback ConnectionCallback DataCallback ErrorCallback ListenerErrorCallback WritableCallback
struct AcceptSlot AddrinfoRange TcpConnection TcpConnectionConfig TcpListener TcpListenerConfig
class TcpConnectionState TcpListenerState
fn addrinfoRange applyKeepAlive applySocketBufferSizes copyStringz mapAdoptError mapConnectError mapListenError mapWsaError portToString queryLocalPort setNonBlocking setTcpNoDelay socketError validated wouldBlock

Types 16

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.
int recvBufferSizeSocket receive buffer size in bytes. 0 = use OS default.
int sendBufferSizeSocket send buffer size in bytes. 0 = use OS default.
bool keepAliveEnabledEnable TCP keepalive probes.
uint keepAliveIdleTime in milliseconds before the first keepalive probe. Default: 60s.
uint keepAliveIntervalTime in milliseconds between keepalive probes. Default: 10s.
Methods
bool isValid() @property const pure @safe nothrow @nogcValidate the config.
structTcpListenerConfig

Listener-side socket configuration.

Fields
int backlogPending accept backlog passed to `listen`.
int concurrentAcceptsNumber of concurrent `AcceptEx` operations to keep pending.
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
private aliasListenerErrorCallback = void delegate(ref TcpListener listener, int errorNumber) @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 @trusted Set the connect callback. Fires once when the TLS/TCP handshake completes successfully. Thread affinity: event loop thread only. Non-reentrant. Does not fire after `dispose()`. Params: ...
void onData(DataCallback callback) @property @trusted Set the inbound data callback. Fires when data arrives. The `data` parameter is borrowed — valid only for the duration of the callback. Copy it if needed beyond. Thread affinity: event loop ...
void onWritable(WritableCallback callback) @property @trusted Set the writable callback. Fires when the send queue drains below the low water mark after backpressure. Not called on initial connect. Thread affinity: event loop thread only. Non-reentrant. ...
void onClose(CloseCallback callback) @property @trusted Set the close callback. Fires when the remote end closes the connection or a fatal error occurs. May fire after `close()` if data was still pending. Does not fire after `dispose()`. Thread af...
void onError(ErrorCallback callback) @property @trusted Set the error callback. Fires on connection errors. After this callback, the connection enters an error state; `onClose` may or may not follow. Thread affinity: event loop thread only. Non-ree...
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.
SendResult sendFile(HANDLE fileHandle, ulong offset, size_t count) @trusted nothrowSend a file over the connection using zero-copy TransmitFile.
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 @trusted Set the accept callback. Fires when a new client connection is accepted. The `clientHandle` is *caller-owned — you must call `TcpConnection.adopt()` to take ownership, or close it via `close...
void onConnection(ConnectionCallback callback) @property @trustedSet the high-level connection callback.
void onError(ListenerErrorCallback callback) @property @trustedSet the error callback for accept failures.
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 SOCKET _socket
private Token _token
private TcpConnectionConfig _config
private bool _readingPaused
private bool _watchingWrite
private bool _awaitingWritable
private bool _closeDelivered
private ubyte[] _sendQueue
private SocketState _state
private int _lastError
private IpAddress _remoteAddress
private IpAddress _localAddress
private CancelToken _cancelToken
private CancelToken _shutdownToken
private bool _draining
private bool _useOverlappedIoWhether this connection uses overlapped I/O mode.
private Token _prepareTokenPrepare token for polling IOCP completion status.
private OVERLAPPED _readOverlappedOverlapped structure for pending read operations.
private OVERLAPPED _writeOverlappedOverlapped structure for pending write operations.
private ubyte[TCP_READ_BUFFER_SIZE] _readBufferRead buffer — must remain valid during overlapped read.
private ubyte[TCP_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 size_t _pendingWriteLenNumber of bytes in the current pending write.
private ConnectExFunc _connectExConnectEx function pointer (loaded on first use).
private OVERLAPPED _connectOverlappedOverlapped structure for pending connect operation.
private bool _connectInProgressWhether a connect operation is currently in progress.
private ubyte[] _pausedReceiveBufferBuffer for data received while reading is paused (IOCP mode only). When paused, incoming data is stored here instead of being delivered.
private TransmitFileFunc _transmitFileFuncTransmitFile function pointer (loaded on first use).
private HANDLE _sendFileHandleFile handle for in-progress file transfer.
private OVERLAPPED _sendFileOverlappedOverlapped structure for pending TransmitFile operation.
private ulong _sendFileOffsetCurrent byte offset in the file being transferred.
private size_t _sendFileRemainingRemaining bytes to transfer (0 = send until EOF).
private bool _sendFileInProgressWhether a TransmitFile operation is currently in progress.
private bool _sendFileCloseAfterWhether to close the file handle after transfer completes.
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
private int bindToAnyAddress(SOCKET sock, int family) @trusted nothrowBind socket to any available local address.
private int submitConnectEx(const(sockaddr) * addr, int addrlen) @trusted nothrowSubmit a ConnectEx operation.
SendResult send(scope const(ubyte)[] data) @trusted nothrowSend data on the connection.
SendResult sendFile(HANDLE fileHandle, ulong offset, size_t count) @trusted nothrowSend a file over the connection using zero-copy TransmitFile.
bool submitTransmitFile() @trusted nothrowSubmit a TransmitFile operation (IOCP overlapped mode).
void processTransmitFileCompletion() @trusted nothrowProcess a completed TransmitFile operation (IOCP overlapped mode).
void doSendFilePoll() @trusted nothrowPerform file send in WSAPoll mode using read + send.
void finishFileTransfer(bool error) @trusted nothrowClean up after a file transfer completes or fails.
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
private void handlePrepare(ref EventLoop loop, Token token) @trusted nothrowHandle prepare phase callback from the event loop.
private void processConnectCompletion() @trusted nothrowProcess a completed ConnectEx operation.
private int associateWithIocp() @trusted nothrowAssociate the socket with the event loop's IOCP.
private void submitRecv() @trusted nothrow Submit an overlapped receive operation. Initiates an asynchronous receive on the socket. The operation will complete via IOCP, and the prepare handler will process the result. * Only one receiv...
private void processRecvCompletion() @trusted nothrowProcess a completed receive operation.
private bool submitSend() @trusted nothrowSubmit an overlapped send operation.
private void processSendCompletion() @trusted nothrowProcess a completed send operation.
private void consumeSendQueue(size_t count) @trusted nothrow @nogcRemove `count` bytes from the front of `sendQueue` without allocating.
Constructors
this(TcpConnectionConfig config)
private classTcpListenerState : IShutdownTrackable
Fields
AcceptCallback onAccept
ConnectionCallback onConnection
ListenerErrorCallback onAcceptError
private EventLoop * _loop
private SOCKET _socket
private Token _token
private TcpListenerConfig _config
private ushort _localPort
private int _lastError
private CancelToken _shutdownToken
private bool _draining
private bool _useOverlappedIoWhether this listener uses overlapped I/O mode (AcceptEx).
private Token _prepareTokenPrepare token for polling IOCP completion status.
private AcceptExFunc _acceptExAcceptEx function pointer.
private GetAcceptExSockaddrsFunc _getAcceptExSockaddrsGetAcceptExSockaddrs function pointer.
private int _addressFamilyAddress family of the listen socket (AF_INET or AF_INET6).
private AcceptSlot[] _acceptSlotsArray of concurrent accept operation slots.
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 invokeAccept(Handle clientHandle) @trusted nothrow
private void invokeAcceptError(int errorNumber) @trusted nothrow
private void handlePrepare(ref EventLoop loop, Token token) @trusted nothrowHandle prepare phase callback from the event loop.
private int associateWithIocp() @trusted nothrowAssociate the listen socket with the event loop's IOCP.
private void submitAccept() @trusted nothrowSubmit AcceptEx operations to fill all available slots.
private void submitAcceptToSlot(ref AcceptSlot slot) @trusted nothrowSubmit an AcceptEx operation to a specific slot.
private void processAcceptCompletion(ref AcceptSlot slot) @trusted nothrowProcess a completed AcceptEx operation for a specific slot.
Constructors
this(TcpListenerConfig config)
private structAcceptSlot

Per-accept operation state for concurrent AcceptEx.

Each slot maintains independent state for one pending AcceptEx call, allowing multiple accepts to be in flight simultaneously.

Fields
SOCKET acceptSocketPre-created socket for this pending accept operation.
OVERLAPPED overlappedOverlapped structure for this pending accept operation.
ubyte[ACCEPT_BUFFER_SIZE] bufferBuffer for AcceptEx output (local + remote addresses).
bool inProgressWhether an accept operation is currently in progress in this slot.
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 16

private fnListenResult mapListenError(int err) pure @safe nothrow @nogcMap Winsock error code to ListenResult.
private fnConnectResult mapConnectError(int err) pure @safe nothrow @nogcMap Winsock error code to ConnectResult.
private fnAdoptResult mapAdoptError(int err) pure @safe nothrow @nogcMap Winsock error code to AdoptResult.
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, scope char[] buffer) @trusted nothrowCopy a D string slice to a null-terminated C string buffer.
private fnconst(char) * portToString(ushort port, scope char[] buffer) @trusted nothrowConvert a port number to a null-terminated string.
private fnint setNonBlocking(SOCKET sock) @trusted nothrowSet a socket to non-blocking mode.
private fnint setTcpNoDelay(SOCKET sock) @trusted nothrow @nogcDisable Nagle's algorithm on a socket.
private fnvoid applySocketBufferSizes(SOCKET sock, const ref TcpConnectionConfig config) @trusted nothrowApply socket buffer size options from config.
private fnvoid applyKeepAlive(SOCKET sock, const ref TcpConnectionConfig config) @trusted nothrowEnable TCP keepalive with configured intervals using WSAIoctl.
private fnbool wouldBlock() @trusted nothrowCheck if the last Winsock error indicates the operation would block.
private fnint socketError(SOCKET sock) @trusted nothrowQuery the socket error status.
private fnint mapWsaError(int wsaError) pure @safe nothrow @nogcMap Winsock error code to POSIX-style errno.
private fnushort queryLocalPort(SOCKET sock) @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.

Variables 4

private enumvarSOMAXCONN = 0x7fffffff

Maximum listen backlog (not in druntime's winsock2).

private enumvarTCP_READ_BUFFER_SIZE = 8192

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

private enumvarACCEPT_ADDR_LEN = SOCKADDR_STORAGE.sizeof + 16

Size of address buffer for AcceptEx (sockaddr_storage + 16 bytes padding).

private enumvarACCEPT_BUFFER_SIZE = ACCEPT_ADDR_LEN * 2

Total buffer size for AcceptEx (local + remote addresses).