//! TCP connections, and listeners that accept them.
//!
//! `connect` opens a `TcpStream` to a `"host:port"` address, and `bind` opens
//! a `TcpListener` whose `accept` waits for connections from other programs.
//! A `TcpStream` reads and writes through `std/io`'s `Read` and `Write`, and
//! failures are `std/io`'s `IoError`s. Every module sees these names without
//! an import, and a module's own definition of one of them shadows the one
//! here.
//!
//! The operations that wait suspend only the calling task: `connect`,
//! `accept`, and a stream's `read`, `write` and `read_timeout` park the task
//! until the socket is ready, and other tasks run meanwhile. A host name,
//! where a numeric address could be given, is looked up first, and the lookup
//! blocks every task until it finishes.
//!
//! A connection over the loopback address: a listener on a port the system
//! picks, a client that connects to it, and a message read to its end:
//!
//! ```
//! async fn main() {
//! let listener = bind("127.0.0.1:0").unwrap() // port 0: the system picks one
//! let addr = "127.0.0.1:${listener.local_port().unwrap()}"
//! let client = connect(addr).await.unwrap()
//! let server = listener.accept().await.unwrap()
//!
//! client.write(bytes_from_string("ping")).await.unwrap()
//! client.close().await.unwrap() // the server's reads now reach the end
//! let got = server.read_to_end().await.unwrap()
//! assert_eq(got.to_string().unwrap(), "ping")
//!
//! server.close().await.unwrap()
//! listener.close().await.unwrap()
//! }
//! ```
// Nova standard library — networking.
//
// Compiled as an implicit module and glob-imported into every user module, so
// these names need no `import` (see docs/adr/0004-stdlib-compile-model.md).
// Error types (`IoError`/`IoErrorKind`) and the `Read`/`Write` traits live in
// `std/io`; `TcpStream` implements both here, the same split `std/fs`'s
// `File` already follows.
//
// **This module's operations genuinely suspend.** Unlike `std/fs`, whose own
// module doc comment states plainly that none of its `async fn`s suspend --
// and not because no poller exists, but because a regular file is not
// readiness-pollable by `select`/`WSAPoll` on any of this project's three CI
// platforms (docs/adr/0013-io-poller.md's IOCP alternative) -- `connect`,
// `read`, `write`, `read_timeout` and `TcpListener::accept` below run against
// `crates/nova-runtime/src/net.rs`'s
// non-blocking sockets and this crate's real poller
// (`crates/nova-runtime/src/poll.rs`): a call that cannot complete
// immediately parks the calling task and lets a sibling task run in the
// meantime, rather than blocking the whole executor's thread. See
// `tests/runtime/net_interleave.nova` for the fixture that actually proves
// this end to end — a merely-correct round trip cannot, because a fully
// blocking implementation underneath this exact same Nova-level surface
// would produce the identical bytes.
//
// **The `Future<T>` spelling.** `async fn` in a trait *declaration* is
// `E0900`, exactly as `std/io`'s own header explains — so `Read`/`Write`
// below are implemented as plain, non-`async` `fn`s returning `Future<T>`,
// each one calling a private `async fn` and returning its future unawaited.
//
// **Every wrapper below awaits its own future-constructor builtin (`net_connect`/
// `net_read`/`net_write`/`net_read_timeout`/`net_accept`) exactly once, then
// reads the per-task payload/message slot with no further `.await` in
// between.** Those
// slots are per-task, keyed on the polling task
// (`crates/nova-runtime/src/fs.rs`'s `SLOTS`/`slot_index`), so a sibling task
// cannot clobber them — but within one task a second operation still
// overwrites that same task's own slot, so a wrapper that awaited again
// before reading its result could read back a payload its own later call had
// already replaced. Keep them straight-line, the same discipline `std/fs`
// and `std/io` already document for their own wrappers.
//
// `net_close`, `net_listen` and `net_local_port` are the exceptions, exactly
// as `crates/nova-resolver/src/lib.rs`'s own doc comments for
// `Builtin::NetClose`, `Builtin::NetListen` and `Builtin::NetLocalPort` state:
// all three are ordinary status words rather than future constructors, so
// `TcpStream::close`, `TcpListener::close`, `bind` and
// `TcpListener::local_port` below call them directly, with no `.await` at all
// — the identical shape `std/fs`'s `File::close` already uses for
// `file_close`. The straight-line rule still applies to them: `bind` and
// `local_port` each read their payload out of the slot with nothing awaited in
// between.
/// An open TCP connection, from `connect` or `TcpListener::accept`.
///
/// It implements `std/io`'s `Read` and `Write`. A read returns empty bytes
/// once the peer has closed the connection, a write can write fewer bytes
/// than it is given, and `flush` does nothing. Copying a `TcpStream` copies
/// the handle: every copy refers to the same connection. Nothing closes it
/// for you: call `close` when done, or it stays open until the program exits.
pub record TcpStream { fd: Int }
// Connect to `addr` ("host:port"), suspending the calling task until the
// connection settles (or fails) rather than blocking every other task on
// this thread for the round trip.
/// Opens a TCP connection to `addr`, a `"host:port"` string such as
/// `"127.0.0.1:8080"`.
///
/// An `async fn`: it suspends the calling task until the connection is made
/// or fails, and other tasks run meanwhile. A host name is looked up first,
/// blocking every task, and only the first address found is tried. Returns an
/// `IoError` of kind `ConnectionRefused` if nothing is listening at `addr`,
/// and `Other` if `addr` is not a valid address. Abandoning a `connect` under
/// `std/time`'s `timeout` leaves its socket open until the program exits.
pub async fn connect(addr: String) -> Result<TcpStream, IoError> {
let status = net_connect(addr).await
if status == 0 { return Ok(TcpStream { fd: decode_count(fs_take_bytes()) }) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
impl TcpStream {
// Closes the connection. Idempotent: a second or third call also
// returns `Ok(())`, because Nova has no move checking, so `self` by
// value cannot stop a second call — the identical reasoning `std/fs`'s
// `File::close` documents. Closing a stale or forged `TcpStream` is the
// same no-op: the runtime's socket table lookup simply removes nothing.
/// Closes the connection, so the peer's reads reach the end of the stream.
///
/// It returns `Ok(())`, also when the stream is already closed: closing
/// reports no error. Afterwards, `read`, `write` and `read_timeout` on the
/// stream, through any copy, return an `IoError` of kind `Other`.
///
/// An `async fn` that never suspends.
pub async fn close(self) -> Result<(), IoError> {
let status = net_close(self.fd)
if status == 0 { return Ok(()) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
// Reads up to `max` bytes, reporting `TimedOut` if `ms` milliseconds
// pass with nothing to read first. Otherwise identical to `Read::read`
// below, including EOF/short-read semantics: an **empty** result is end
// of stream, and a short read is not.
//
// No `std/fs` analogue exists (no timeout-based read there), so this is
// modelled as a second inherent method on `TcpStream`, parallel to
// `close` above, rather than folded into the `Read` trait, which has no
// room for a third argument.
/// Reads up to `max` bytes, as `Read::read` does, but gives up if `ms`
/// milliseconds pass with nothing to read.
///
/// An `async fn`: it suspends the calling task while it waits, and other
/// tasks run meanwhile. Empty bytes mean the peer has closed the
/// connection. Returns an `IoError` of kind `TimedOut` if the time runs
/// out; with `ms` 0 or negative that happens at once, unless bytes or the
/// end of the stream are already there. On a closed stream it returns one
/// of kind `Other`. Panics if `max` is negative.
pub async fn read_timeout(self, max: Int, ms: Int) -> Result<Bytes, IoError> {
let status = net_read_timeout(self.fd, max, ms).await
if status == 0 { return Ok(fs_take_bytes()) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
}
// Any operation other than `close` against an `fd` the runtime's socket
// table does not recognise — already closed, stale, or forged (`TcpStream`'s
// field is not privacy-enforced, so `TcpStream { fd: 9999 }` compiles and
// names no connection this module ever opened) — reaches this same
// `IoError { kind: Other }` by missing the same table lookup, whether the
// operation is this function, `write_stream` below, or `read_timeout` above.
async fn read_stream(s: TcpStream, max: Int) -> Result<Bytes, IoError> {
let status = net_read(s.fd, max).await
if status == 0 { return Ok(fs_take_bytes()) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
impl Read for TcpStream {
fn read(self, max: Int) -> Future<Result<Bytes, IoError>> { read_stream(self, max) }
}
// The write's byte count crosses the boundary the same way `std/fs`'s
// `write_file` decodes its own: 8 little-endian bytes in the payload slot,
// read back here by that same shared `decode_count`.
//
// **May write fewer bytes than given** — one non-blocking `Write::write`
// attempt, not a `write_all` loop, per `net_write`'s own doc comment
// (`crates/nova-runtime/src/net.rs`). Mirrors `std/io`'s `Write::write`
// contract exactly, deliberately unlike `std/fs`'s count-less, write-all
// `write`: a caller here that needs a guaranteed full write must loop on the
// returned count itself.
async fn write_stream(s: TcpStream, buf: Bytes) -> Result<Int, IoError> {
let status = net_write(s.fd, buf).await
if status == 0 { return Ok(decode_count(fs_take_bytes())) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
// Always `Ok(())`, immediately, with no builtin call at all.
//
// **Deliberate, not a stub** — the same reasoning `std/io`'s own
// `flush_stderr` doc comment gives for a status check it calls "provably
// unpinnable": one `write` here is already one unbuffered syscall (`net_write`
// wraps a single `std::net::TcpStream::write` attempt), so there is no
// userspace buffer for a flush to push out, exactly the shape Rust's own
// `impl Write for std::net::TcpStream` uses for the identical reason. There
// is no `net_flush` builtin and none is added to back a no-op. (This comment
// used to say "a sixth one", counting the five net builtins that existed when
// it was written; the count was stale within one increment, so it is gone
// rather than maintained.) **If `std/net` ever grows write buffering, this
// becomes load-bearing and needs a real intrinsic** — a future reader should
// not mistake this for an oversight.
async fn flush_stream(s: TcpStream) -> Result<(), IoError> { Ok(()) }
impl Write for TcpStream {
fn write(self, buf: Bytes) -> Future<Result<Int, IoError>> { write_stream(self, buf) }
fn flush(self) -> Future<Result<(), IoError>> { flush_stream(self) }
}
// A listening socket. Server side of `std/net`, added because Phase 2
// position 10 (`std/http`) has no transport without it.
/// A TCP socket that listens for connections, from `bind`.
///
/// `accept` waits for the next connection and returns a `TcpStream` for it.
/// Nothing closes a listener for you: call `close` when done, or it holds its
/// port until the program exits.
pub record TcpListener { fd: Int }
// Bind and listen on `addr` ("host:port"). Pass port 0 to let the kernel
// choose, then read it back with `local_port`.
//
// **A plain `fn`, not an `async fn`: this cannot suspend.** The `bind` and
// `listen` syscalls do not block, so `net_listen` is an ordinary status word
// and there is no future to await. Forcing `.await` here would misstate the
// cost. Note the deliberate exception at `TcpListener::close` below.
//
// **One caveat, and it is about `addr` rather than about the syscalls.** A
// numeric address ("127.0.0.1:8080") never blocks. A *hostname* is resolved
// by the runtime through Rust's `ToSocketAddrs`, which is a blocking DNS
// lookup with no future to park on -- so passing one blocks this whole
// executor thread, every sibling task included, for as long as resolution
// takes. `connect` above carries the identical caveat for the identical
// reason (`resolve_addr` in `crates/nova-runtime/src/net.rs`); neither
// function makes name resolution asynchronous, and this module does not
// pretend otherwise. Pass a numeric address on any path where that matters.
// The two also resolve by *different* policies for a name yielding several
// addresses -- see `nova_rt_net_listen`'s own doc comment.
//
// An address already in use arrives as `IoError { kind: Other }`, not a
// distinct kind: `IoErrorKind` has eight variants and `AddrInUse` is not
// among them, so the most likely failure of this function is discriminable
// only by `message`, whose text is the operating system's own. Adding a
// variant is a wire-contract change across the runtime's `fail` and
// `std/io`'s `io_error_kind_of` together, and is deliberately not done here.
/// Opens a listener on `addr`, a `"host:port"` string such as
/// `"127.0.0.1:8080"`.
///
/// Port 0 lets the system choose a free port, which `local_port` then
/// reports. It is not `async`: binding never waits for a peer, though a host
/// name is looked up first, blocking every task. Returns an `IoError` of kind
/// `Other` if the address is already in use, which only the error's
/// `message` tells apart, and also if `addr` is not a valid address.
pub fn bind(addr: String) -> Result<TcpListener, IoError> {
let status = net_listen(addr)
if status == 0 { return Ok(TcpListener { fd: decode_count(fs_take_bytes()) }) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
impl TcpListener {
// Stops accepting and releases the port. Idempotent, for the same reason
// `TcpStream::close` is: Nova has no move checking, so `self` by value
// cannot prevent a second call.
//
// **An `async fn` even though it cannot suspend** -- the one deliberate
// exception to the rule stated at `bind` above. It calls the same
// `net_close` a stream does, with no `.await`, because the runtime's
// table removes an entry by key regardless of which kind it holds. It is
// declared `async` to match the shipped `TcpStream::close`: a server
// closes both handles in sequence, and a reader should not have to
// remember which of two identically named socket methods needs `.await`.
/// Closes the listener and frees its port.
///
/// It returns `Ok(())`, also when the listener is already closed: closing
/// reports no error. An `accept` already waiting on the listener goes on
/// waiting, and keeps the program from ending, so close a listener only
/// when no `accept` is waiting on it.
///
/// An `async fn`, like `TcpStream::close`, that never suspends.
pub async fn close(self) -> Result<(), IoError> {
let status = net_close(self.fd)
if status == 0 { return Ok(()) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
// The port this listener is bound to, which is how a caller learns the
// kernel's choice after binding port 0.
//
// **A plain `fn`, not an `async fn`: this cannot suspend** -- the same
// rule `bind` states, and the same one `close` above deliberately
// excepts. Reading a bound port is bookkeeping the kernel already holds,
// and unlike `bind` there is no address argument at all, so `bind`'s
// hostname caveat has nothing to apply to here: this one cannot block on
// any argument.
//
// The port arrives in the payload slot rather than as the returned status,
// because that status *is* the error kind -- a port carried there would
// make `0` ambiguous and every non-zero port indistinguishable from a
// failure. So it is read back with the same `decode_count` `bind` uses for
// its fd, straight-line with nothing awaited in between.
//
// Errors as `IoError { kind: Other }` on a closed, stale or forged
// listener, by the same missed table lookup every other operation reports
// that way.
//
// **A `TcpListener` whose `fd` actually names a connected stream is that
// same case, and is not distinguishable from it.** `fd` is not
// privacy-enforced, so `TcpListener { fd: s.fd }` built from a `TcpStream`
// compiles; the runtime keeps streams and listeners in one table and its
// kind-checked lookup collapses wrong-kind into the identical miss an
// absent fd gives. `IoErrorKind` has no wrong-kind variant to report
// instead -- the same reason `bind` above gives for `AddrInUse` arriving as
// `Other` -- so a stream fd here yields `Other` with the runtime's own
// "socket is not open" message, and no Nova code can tell the two apart.
/// Returns the port the listener is bound to: after binding port 0, the
/// one the system chose.
///
/// It is not `async`, and never waits. Returns an `IoError` of kind
/// `Other` if the listener is closed.
pub fn local_port(self) -> Result<Int, IoError> {
let status = net_local_port(self.fd)
if status == 0 { return Ok(decode_count(fs_take_bytes())) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
// Waits for the next incoming connection, suspending the calling task
// rather than blocking the executor.
//
// **An `async fn` on its merits, not by convention** -- the rule `bind`
// states is "plain `fn` for anything that cannot suspend", and this is the
// one operation on a listener that genuinely can. The `accept` syscall
// waits on a peer, so `net_accept` is a future constructor and the
// `.await` here is real: while it is parked, sibling tasks on this thread
// keep running.
//
// The returned `TcpStream` is indistinguishable from a connected one, so
// `Read`, `Write`, `read_timeout` and `close` above all work on it
// unchanged. Its fd arrives in the payload slot rather than as the
// returned status, read back with the same `decode_count` `bind` and
// `connect` use for their own -- straight-line, with nothing awaited
// between the `.await` above and the read, the discipline this module's
// header states.
//
// **One socket wait per task, enforced by process abort.** Staging two
// I/O waits in a single poll ends the process (`stage_park` in
// `crates/nova-runtime/src/task.rs`), so a task parked in `accept` cannot
// also be reading or writing a connection: a server needs at least two
// tasks -- one accepting, one per connection it serves -- and this
// language has no `select` or `race` to fold them into one. `spawn` is the
// whole of what makes that possible.
//
// **No timeout.** There is no `accept_timeout` counterpart to
// `TcpStream::read_timeout`, so a task waiting here waits until a
// connection arrives or the process ends; an untimed wait is never
// reported as a deadlock, and `block_on` cannot return while any task is
// parked. A graceful shutdown therefore polls this under `std/time`'s
// `timeout` instead of awaiting it bare, and closes the listener only
// once the abandoned accept has returned: closing it under a parked
// accept does not wake that accept. `std/http`'s `Server::listen` is that
// shape.
//
// Errors as `IoError { kind: Other }` on a closed, stale or forged
// listener, and on a `TcpListener` whose `fd` actually names a connected
// stream -- the same wrong-kind collapse `local_port` above documents at
// length, and not distinguishable from the closed case. Both halves of
// that promise are measured, in `net.rs`'s
// `accepting_on_a_closed_or_wrong_kind_fd_is_an_error_not_a_park`; until
// that test was added this comment asserted a boundary nothing checked.
/// Waits for the next incoming connection, and returns a `TcpStream` for
/// it.
///
/// An `async fn`: it suspends the calling task until a connection
/// arrives, and other tasks run meanwhile. It has no time limit, and the
/// program does not end while it waits; closing the listener does not end
/// the wait either. To stop waiting, run it under `std/time`'s `timeout`.
/// A task waiting here can do nothing else, so a server accepts in one
/// task and serves each connection in another, started with `spawn`.
/// Returns an `IoError` of kind `Other` if the listener is closed.
///
/// Here the server accepts and serves in a task of its own, while this
/// one runs the client:
///
/// ```
/// async fn serve_one(listener: TcpListener) {
/// let conn = listener.accept().await.unwrap()
/// let mut asked = bytes_from_string("")
/// while asked.len() < 4 { // a read may return fewer bytes than were sent
/// asked = asked.concat(conn.read(64).await.unwrap())
/// }
/// assert_eq(asked.to_string().unwrap(), "ping")
/// conn.write(bytes_from_string("pong")).await.unwrap()
/// conn.close().await.unwrap()
/// }
///
/// async fn main() {
/// let listener = bind("127.0.0.1:0").unwrap()
/// let port = listener.local_port().unwrap()
/// let server = spawn(serve_one(listener)) // accept in a task of its own
/// let client = connect("127.0.0.1:${port}").await.unwrap()
/// client.write(bytes_from_string("ping")).await.unwrap()
/// let reply = client.read_to_end().await.unwrap() // until the server closes
/// assert_eq(reply.to_string().unwrap(), "pong")
/// server.join().await
/// listener.close().await.unwrap()
/// }
/// ```
pub async fn accept(self) -> Result<TcpStream, IoError> {
let status = net_accept(self.fd).await
if status == 0 { return Ok(TcpStream { fd: decode_count(fs_take_bytes()) }) }
Err(IoError { kind: io_error_kind_of(status), message: fs_last_error_message() })
}
}