# dista.conn

> A module of Dista's API.

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`

```mojo
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__`

```mojo
def __init__(out self, var stream: Stream)
```

#### `fd`

```mojo
def fd(self) -> c_int
```

#### `read_available`

```mojo
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`

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

The next complete frame, if there is one.

#### `send`

```mojo
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`

```mojo
def finish_connect(mut self)
```

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

#### `send_parts`

```mojo
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`

```mojo
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`

```mojo
def flush(mut self)
```

Sends queued output until the socket would block.

#### `release_buffers`

```mojo
def release_buffers(mut self)
```

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

#### `wants_write`

```mojo
def wants_write(self) -> Bool
```

#### `unsent`

```mojo
def unsent(self) -> Int
```

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

## Aliases

### `BIG_FRAME`

```mojo
comptime BIG_FRAME = 262144
```

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