asyncio.vek

// std:asyncio — reactor-backed async I/O readiness + the descriptors it works over.
//
// The low-level layer higher-level I/O (sockets, pipes) builds on: create and
// configure descriptors, and — the point of this module — wait for a descriptor
// to become readable or writable *without blocking the worker thread* a task runs
// on. A wait parks the calling task and frees its worker (§10 "Asynchronous I/O");
// off a task it blocks only that kernel thread. The pattern is: set the descriptor
// non-blocking, attempt the syscall, and on `EAGAIN` wait for readiness and retry.
//
// A returned wait is a hint that the descriptor is probably ready — the following
// syscall observes the true state (another task may have consumed it), so callers
// always retry the syscall rather than assume data is present.

// ---- runtime bindings ----------------------------------------------------

native fn __rt_os_pipe(out: i32[]) -> i32;
native fn __rt_os_set_nonblocking(fd: i32) -> i32;
native fn __rt_reactor_wait_readable(fd: i32) -> i32;
native fn __rt_reactor_wait_writable(fd: i32) -> i32;
native fn __rt_reactor_arm(fd: i32, events: i32, edge: bool) -> i32;
native fn __rt_reactor_wait(fd: i32, events: i32) -> i32;
native fn __rt_reactor_disarm(fd: i32) -> i32;
// Raw descriptor I/O (shared with std:fs — these operate on any fd, not just
// files), so an `os` caller need not also import std:fs for a pipe or socket.
native fn __rt_fs_read(fd: i32, buf: u8[], off: usize, len: usize) -> i64;
native fn __rt_fs_write(fd: i32, buf: u8[], off: usize, len: usize) -> i64;

// ---- readiness event mask ------------------------------------------------

// Passed to `arm`/`wait`; also the mask `wait` returns. Bitwise-or for both.
pub const READABLE: i32 = 1;
pub const WRITABLE: i32 = 2;

// ---- descriptors ---------------------------------------------------------

// Create an anonymous pipe. Returns `(read_end, write_end)`; on failure both are
// the negative errno.
pub fn pipe() -> (i32, i32) {
  let fds: i32[] = [0, 0];
  let rc = __rt_os_pipe(fds);
  if rc < 0 {
    return (rc, rc);
  }
  return (fds[0], fds[1]);
}

// Put `fd` in non-blocking mode (`O_NONBLOCK`). Returns 0, or a negative errno.
pub fn set_nonblocking(fd: i32) -> i32 {
  return __rt_os_set_nonblocking(fd);
}

// Read into `buf`; returns the byte count, 0 at end of input, or a negative errno
// (`-EAGAIN` when a non-blocking descriptor has nothing ready).
pub fn read(fd: i32, buf: u8[]) -> i64 {
  return __rt_fs_read(fd, buf, 0, buf.length() as usize);
}

// Write `buf`; returns the byte count written, or a negative errno.
pub fn write(fd: i32, buf: u8[]) -> i64 {
  return __rt_fs_write(fd, buf, 0, buf.length() as usize);
}

// ---- readiness (level-triggered, one-shot) -------------------------------

// Suspend until `fd` is readable, then return 0 (or a negative errno). Safe to
// call at any time — returns immediately if `fd` is already readable.
pub fn wait_readable(fd: i32) -> i32 {
  return __rt_reactor_wait_readable(fd);
}

// Suspend until `fd` is writable, then return 0 (or a negative errno).
pub fn wait_writable(fd: i32) -> i32 {
  return __rt_reactor_wait_writable(fd);
}

// ---- readiness (persistent; the throughput path) -------------------------

// Register `fd` for persistent readiness notification on `events` (a `READABLE`/
// `WRITABLE` mask). `edge` selects edge-triggered delivery — the caller must then
// drain the descriptor to `EAGAIN` before the next `wait`, in exchange for arming
// the descriptor only once. Returns 0, or a negative errno.
pub fn arm(fd: i32, events: i32, edge: bool) -> i32 {
  return __rt_reactor_arm(fd, events, edge);
}

// Suspend until an armed `fd` signals one of `events`; returns the ready mask
// (a positive `READABLE`/`WRITABLE` combination) or a negative errno.
pub fn wait(fd: i32, events: i32) -> i32 {
  return __rt_reactor_wait(fd, events);
}

// Deregister an armed `fd`. Call before closing it. Returns 0, or a negative errno.
pub fn disarm(fd: i32) -> i32 {
  return __rt_reactor_disarm(fd);
}