morca Dista

Dista · API

dista.conn

A non-blocking framed connection, for the head's event loop.

The head talks to every driver and worker at once, so it must never block on one of them. A Conn reads and writes with MSG_DONTWAIT, so no call blocks. It reads whatever has arrived into an input buffer and hands out complete frames, and queues outgoing frames in an output buffer that flush drains as the socket accepts it. The event loop polls for writability only while something is queued.

A large frame isn't assembled in the input buffer. Once its header has arrived, its payload gets a buffer of its own, of the size the header gives, and the socket is read straight into that: the payload is copied once, out of the kernel, and handed out as the frame's without another copy. Frames that arrived before it are handed out first, and those after it once it's whole.

Structs

Conn

struct Conn

Implements Movable

Fields
stream: Stream
closed: Bool: The peer hung up, or the connection failed.
connecting: Bool: Started with Stream.start_connect and not yet connected: output queues until it is.
deadline: UInt64: While connecting: when to give up, in wall_ns; 0 for never.
why: String: Why the connection failed, if it's known.

__init__

def __init__(out self, var stream: Stream)

fd

def fd(self) -> c_int

read_available

def read_available(mut self, limit: Int = Int(0))

Reads what's waiting on the socket into the input buffer, or a large frame's own: all of it, or with limit, about that many bytes, leaving the rest in the socket, where poll still reports it.

next_frame

def next_frame(mut self) -> Optional[Frame]

The next complete frame, if there is one.

send

def send(mut self, kind: UInt8, object_id: UInt64, task_id: UInt64, payload: Span[UInt8])

Queues a frame and sends as much as the socket takes now.

finish_connect

def finish_connect(mut self)

The socket polled writable while connecting: connected, or raises why not.

send_parts

def send_parts(mut self, object_id: UInt64, data: Span[UInt8], part: Int)

Queues object object_id as part frames of at most part bytes, and sends as much as the socket takes now.

queue

def queue(mut self, data: Span[UInt8])

Adds bytes to the output buffer without sending them: part of a frame whose header went before; flush sends them.

flush

def flush(mut self)

Sends queued output until the socket would block.

release_buffers

def release_buffers(mut self)

Frees both buffers, of a connection that's closed.

wants_write

def wants_write(self) -> Bool

unsent

def unsent(self) -> Int

Bytes queued to send that the peer hasn't taken yet.

Aliases

comptime BIG_FRAME = 262144

Payloads at least this large are read into a buffer of their own.