`dista.socket`

IPv4 TCP and process control over libc.

Just enough of socket(2) and friends for dista's processes: listeners on loopback or on every interface, streams to a host name or to loopback, the non-blocking reads and writes the head's event loop uses, poll, and fork/waitpid/_exit to start and reap local processes. macOS and Linux differ in sockaddr_in's first byte, in addrinfo's field order, in the SOL_SOCKET, MSG_DONTWAIT and EAGAIN constants, and in how a write to a closed peer is kept from raising SIGPIPE; platform_map picks the right value for each.

Structs

`Stream`

struct Stream

A connected TCP socket. It's non-blocking, but write_all and read_into wait for it, so to its users it blocks; the head's loop uses send_some and recv_some, which don't. Closes itself when dropped.

Implements Movable

Fields: fd: c_int

`init`

def __init__(out self, fd: Int32)

`deinit`

def __deinit__(deinit self)

`connect`

def connect(port: Int, timeout_ms: Int = Int(-1)) -> Self

Connects to 127.0.0.1:port, waiting at most timeout_ms (-1: as long as the kernel tries).

def connect(host: String, port: Int, timeout_ms: Int = Int(-1)) -> Self

Connects to host:port; host is a name or a dotted quad. Waits at most timeout_ms (-1: as long as the kernel tries).

def connect(ip: Array[UInt8, Int(4)], port: Int, timeout_ms: Int = Int(-1)) -> Self

Connects to ip:port, waiting at most timeout_ms (-1: as long as the kernel tries).

`start_connect`

def start_connect(ip: Array[UInt8, Int(4)], port: Int) -> Self

Starts connecting to ip:port and returns at once. The socket is writable once the connect completes or fails; finish_connect then says which. An event loop connects this way, so a peer that never answers doesn't stall it.

`finish_connect`

def finish_connect(mut self)

Raises if the connect start_connect began failed. Call it once the socket polls writable (or failed).

`local_ip`

def local_ip(self) -> Array[UInt8, Int(4)]

The local IPv4 address this connection goes out from.

`write_all`

def write_all(self, data: Span[UInt8])

Writes every byte of data.

`read_into`

def read_into(self, buf: Pointer[UInt8], size: Int) -> Int

Reads exactly size bytes into buf.

Returns size, or 0 when the peer closed before sending anything. Raises if the peer closes partway through.

`close`

def close(mut self)

`Listener`

struct Listener

A TCP listener. Closes itself when dropped.

Implements Movable

Fields: fd: c_int; port: Int

`init`

def __init__(out self, port: Int = Int(0), any_interface: Bool = False, backlog: Int = Int(128))

Binds and starts listening.

Args: port: The port, or 0 for one the OS picks.; any_interface: Listen on every interface instead of loopback.; backlog: How many connections may wait to be accepted.

`deinit`

def __deinit__(deinit self)

`accept`

def accept(self) -> Stream

Waits for the next connection.

`close`

def close(mut self)

Functions

`resolve_ipv4`

def resolve_ipv4(host: String) -> Array[UInt8, Int(4)]

The first IPv4 address of host, a name or a dotted quad.

`send_some`

def send_some(fd: Int32, data: Pointer[UInt8], size: Int) -> Int

Sends what the socket takes without blocking. Returns the bytes sent.

`recv_some`

def recv_some(fd: Int32, buf: Pointer[UInt8], size: Int) -> Int

Reads what has arrived, without blocking.

Returns the bytes read, 0 at end of stream, or -1 if nothing is waiting.

`fcntl`

def fcntl(fd: Int32, cmd: Int32, arg: Int32) -> c_int

fcntl(2) with one int argument: the only declaration of it in dista, since two with different signatures don't link.

`set_nonblocking`

def set_nonblocking(fd: Int32) -> Bool

Makes fd non-blocking; whether that worked.

`keepalive_seconds`

def keepalive_seconds() -> Int

DISTA_KEEPALIVE_SECONDS: how long a connection may be silent before the kernel probes it; 0 turns keepalive off. Default 10.

`set_keepalive`

def set_keepalive(fd: Int32)

Has the kernel notice a peer that went silent without closing: a host that froze, lost its network or was hard-stopped, which no TCP close reports. After keepalive_seconds() of silence it probes every half of that, and after KEEPALIVE_PROBES unanswered probes the socket fails, about 25 s by default; poll then reports it as an error, and dista drops it as it does a closed one.

Probes go only while nothing sent is unacknowledged; with data in flight the kernel's retransmission limit applies instead (about 15 minutes on Linux). TCP_USER_TIMEOUT would shorten that, but it also fails a connection whose peer is alive and only not reading, as a driver busy between gets, or one the head holds back for admission, is. DISTA_TASK_TIMEOUT bounds a task however its worker stalls.

`get_int_option`

def get_int_option(fd: Int32, level: Int32, name: Int32) -> Int

A socket option's int value, or -1 if it can't be read.

`socket_error`

def socket_error(fd: Int32) -> c_int

The error pending on fd (SO_ERROR): how a non-blocking connect ended, 0 if it succeeded.

`wall_ns`

def wall_ns() -> UInt64

Nanoseconds since the epoch, from CLOCK_REALTIME.

The timeline stamps tasks on the head and on workers with this, so the stamps compare across processes and, NTP permitting, machines; a monotonic clock only compares within one host.

`stream_pair`

def stream_pair() -> List[Stream]

Two connected local stream sockets, for tests and fuzzing.

`poll`

def poll(fds: List[Int32], events: List[Int16], timeout_ms: Int = Int(-1)) -> List[Int16]

Waits for any of events[i] on fds[i], or until timeout_ms passes.

Returns each descriptor's revents. A timeout of -1 waits indefinitely.

`poll_readable`

def poll_readable(fds: List[Int32], timeout_ms: Int = Int(-1)) -> List[Bool]

Waits until at least one of fds is readable, or timeout_ms passes.

A closed or failed socket counts as readable, so the read that follows reports it. A timeout of -1 waits indefinitely.

`fork`

def fork() -> Int

Forks the process. Returns 0 in the child, the child's pid in the parent.

`wait_child`

def wait_child(pid: Int) -> Int

Waits for child pid to exit and returns its raw wait status.

`exit_now`

def exit_now(code: Int)

Ends the process immediately, without running atexit handlers.

A forked worker leaves through here, so it never returns into the driver's main or flushes the driver's buffered output a second time.

`addr_key`

def addr_key(ip: Array[UInt8, Int(4)], port: Int) -> String

Aliases

comptime AF_INET = Int32(2)

comptime SOCK_STREAM = Int32(1)

comptime SOL_SOCKET = SIMD(platform_map[Optional(String("SOL_SOCKET")), linux=Optional(Int(1)), macos=Optional(Int(65535))]())

comptime SO_REUSEADDR = SIMD(platform_map[Optional(String("SO_REUSEADDR")), linux=Optional(Int(2)), macos=Optional(Int(4))]())

comptime SO_NOSIGPIPE = SIMD(platform_map[Optional(String("SO_NOSIGPIPE")), linux=Optional(Int(0)), macos=Optional(Int(4130))]())

comptime MSG_NOSIGNAL = SIMD(platform_map[Optional(String("MSG_NOSIGNAL")), linux=Optional(Int(16384)), macos=Optional(Int(0))]())

comptime EINTR = Int32(4)

comptime IPPROTO_TCP = Int32(6)

comptime TCP_NODELAY = Int32(1)

comptime SO_KEEPALIVE = SIMD(platform_map[Optional(String("SO_KEEPALIVE")), linux=Optional(Int(9)), macos=Optional(Int(8))]())

comptime TCP_KEEPIDLE = SIMD(platform_map[Optional(String("TCP_KEEPIDLE")), linux=Optional(Int(4)), macos=Optional(Int(16))]())

comptime TCP_KEEPINTVL = SIMD(platform_map[Optional(String("TCP_KEEPINTVL")), linux=Optional(Int(5)), macos=Optional(Int(257))]())

comptime TCP_KEEPCNT = SIMD(platform_map[Optional(String("TCP_KEEPCNT")), linux=Optional(Int(6)), macos=Optional(Int(258))]())

comptime KEEPALIVE_PROBES = 3

comptime SOCKADDR_IN_SIZE = 16

comptime EAGAIN = SIMD(platform_map[Optional(String("EAGAIN")), linux=Optional(Int(11)), macos=Optional(Int(35))]())

comptime EINPROGRESS = SIMD(platform_map[Optional(String("EINPROGRESS")), linux=Optional(Int(115)), macos=Optional(Int(36))]())

comptime SO_ERROR = SIMD(platform_map[Optional(String("SO_ERROR")), linux=Optional(Int(4)), macos=Optional(Int(4103))]())

comptime F_GETFL = Int32(3)

comptime F_SETFL = Int32(4)

comptime O_NONBLOCK = SIMD(platform_map[Optional(String("O_NONBLOCK")), linux=Optional(Int(2048)), macos=Optional(Int(4))]())

comptime MSG_DONTWAIT = SIMD(platform_map[Optional(String("MSG_DONTWAIT")), linux=Optional(Int(64)), macos=Optional(Int(128))]())

comptime ADDRINFO_SIZE = 48

comptime ADDRINFO_FAMILY_OFF = 4

comptime ADDRINFO_SOCKTYPE_OFF = 8

comptime ADDRINFO_ADDR_OFF = platform_map[Optional(String("AI_ADDR_OFF")), linux=Optional(Int(24)), macos=Optional(Int(32))]()

comptime POLLIN = Int16(1)

comptime POLLOUT = Int16(4)

comptime POLLERR = Int16(8)

comptime POLLHUP = Int16(16)

comptime POLLFD_SIZE = 8