Socket

Linear (QTT) Socket library — TCP/UDP via Python FFI. Every socket-consuming operation takes its socket parameter at multiplicity 1 (``(1 s: ...)``), so the type checker enforces the resource discipline at the binder. A ``let 1 s = stream_socket unit in body`` *must* consume ``s`` exactly once: passing it to ``stream_bind``, ``stream_close``, etc. discharges the linear obligation. Forgetting to close any handle, or using one after consumption, is rejected at type-check time. In addition to the linear discipline, refinements track the state-machine: ``stream_socket`` produces a fresh-not-bound-not- connected socket; ``stream_bind`` / ``stream_connect`` advance the state; ``stream_recv`` / ``stream_send`` require connected. Opaque Python wrappers (see native bodies below).
Table of Contents

Types

StreamSocket

(type)
Opaque Python wrappers (see native bodies below).
type StreamSocket

DatagramSocket

(type)
type DatagramSocket

SocketPayload

(type)
Raw payload (`bytes` at runtime).
type SocketPayload

SockAddr

(type)
Runtime: address tuple, e.g. `(host, port)` for `AF_INET`.
type SockAddr

TcpAccept

(type)
Runtime: `(socket, addr)` pair from `TCP accept`.
type TcpAccept

UdpRecvFrom

(type)
Runtime: `(data, addr)` pair from `UDP recvfrom`.
type UdpRecvFrom

Functions

socket

Python `socket` module (binds to the evaluator name `socket`).
def socket : Unit

ipv4_addr

def ipv4_addr (host: String) (port: {p : Int | p ≥ 0 && p ≤ 65535}) : {a : SockAddr | ipv4_sock_addr a = True}

stream_socket

def stream_socket (_: Unit) : {s : StreamSocket | stream_bound s = False && stream_connected s = False}

dgram_socket

def dgram_socket (_: Unit) : {s : DatagramSocket | dgram_bound s = False && dgram_connected s = False}

tcp_accept_socket

def tcp_accept_socket (r: TcpAccept) : {s : StreamSocket | stream_connected s = True}

tcp_accept_peer

def tcp_accept_peer (r: TcpAccept) : {a : SockAddr | ipv4_sock_addr a = True}

stream_bind

def stream_bind (addr: {ad : SockAddr | ipv4_sock_addr ad = True}) (s: {sock : StreamSocket | stream_bound sock = False}) : {out : StreamSocket | stream_bound out = True}

stream_listen

def stream_listen (backlog: {n : Int | n > 0}) (s: {sock : StreamSocket | stream_bound sock = True}) : {out : StreamSocket | stream_bound out = True}

stream_connect

def stream_connect (addr: {ad : SockAddr | ipv4_sock_addr ad = True}) (s: {sock : StreamSocket | stream_connected sock = False}) : {out : StreamSocket | stream_connected out = True}

stream_accept

def stream_accept (s: {sock : StreamSocket | stream_bound sock = True && stream_connected sock = False}) : TcpAccept

stream_send

def stream_send (data: SocketPayload) (s: {sock : StreamSocket | stream_connected sock = True}) : {sent : Int | sent ≥ 0 && sent ≤ payload_len data}

stream_recv

def stream_recv (bufsize: {n : Int | n > 0}) (s: {sock : StreamSocket | stream_connected sock = True}) : {p : SocketPayload | payload_len p ≤ bufsize}

stream_close

``stream_close`` consumes the socket and returns ``Unit`` so the closed socket cannot be used again — it has nowhere to go.
def stream_close (s: StreamSocket) : Unit

dgram_bind

def dgram_bind (addr: {ad : SockAddr | ipv4_sock_addr ad = True}) (s: {sock : DatagramSocket | dgram_bound sock = False}) : {out : DatagramSocket | dgram_bound out = True}

dgram_connect

def dgram_connect (addr: {ad : SockAddr | ipv4_sock_addr ad = True}) (s: DatagramSocket) : {out : DatagramSocket | dgram_connected out = True}

dgram_sendto

def dgram_sendto (data: SocketPayload) (addr: {ad : SockAddr | ipv4_sock_addr ad = True}) (s: DatagramSocket) : {sent : Int | sent ≥ 0 && sent ≤ payload_len data}

dgram_recvfrom

def dgram_recvfrom (bufsize: {n : Int | n > 0}) (s: {sock : DatagramSocket | dgram_bound sock = True}) : {r : UdpRecvFrom | udp_payload_len r ≤ bufsize}

udp_recv_payload

def udp_recv_payload (r: UdpRecvFrom) : {p : SocketPayload | payload_len p = udp_payload_len r}

udp_recv_peer

def udp_recv_peer (r: UdpRecvFrom) : {a : SockAddr | ipv4_sock_addr a = True}

dgram_send

def dgram_send (data: SocketPayload) (s: {sock : DatagramSocket | dgram_connected sock = True}) : {sent : Int | sent ≥ 0 && sent ≤ payload_len data}

dgram_recv

def dgram_recv (bufsize: {n : Int | n > 0}) (s: {sock : DatagramSocket | dgram_connected sock = True}) : {p : SocketPayload | payload_len p ≤ bufsize}

dgram_close

def dgram_close (s: DatagramSocket) : Unit

Uninterpreted

Functions declared as def f ... = uninterpreted: only their signature is known to the verifier; they have no body.

stream_bound

uninterpreted
def stream_bound : (s : StreamSocket) → Bool

stream_connected

uninterpreted
def stream_connected : (s : StreamSocket) → Bool

dgram_bound

uninterpreted
def dgram_bound : (s : DatagramSocket) → Bool

dgram_connected

uninterpreted
def dgram_connected : (s : DatagramSocket) → Bool

payload_len

uninterpreted
Byte length of a payload (`len(data)` at runtime).
def payload_len : (p : SocketPayload) → Int

udp_payload_len

uninterpreted
Length of the datagram payload in a `recvfrom` result.
def udp_payload_len : (r : UdpRecvFrom) → Int

ipv4_sock_addr

uninterpreted
Constructed IPv4 text endpoint (`(host, port)`).
def ipv4_sock_addr : (a : SockAddr) → Bool