feat(io): add fd_read_into — zero-allocation incremental read from an fd

CHANGELOG.md

@@ -9,6 +9,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 `main`, the release pipeline automatically replaces `[current]` with the next
 version number before tagging the release.
 
+## [current]
+
+### Added
+
+- **`io.fd_read_into(fd, buf, length)`** — zero-allocation incremental read
+  from a raw file descriptor (#1471). Mirrors `fs.pread_into`, which was
+  already the same shape for files, so a caller streaming from a pipe now gets
+  the same option as one reading a file.
+
+  Returns `(n, err)`: `n == 0` is EOF, `0 < n < length` is a normal short read.
+  Unlike `io.fd_read_n` — which loops until it has filled the buffer — this
+  returns as soon as *any* bytes are available, which is what makes it usable
+  on a live pipe where waiting to fill would stall until the producer happened
+  to send a whole buffer's worth.
+
+  Motivated by streaming video decode: reading one frame per iteration, an
+  8 MB frame at 1080p30 means ~250 MB/s of allocator churn if every read mints
+  a fresh string. One caller-owned buffer reused across the loop avoids it
+  entirely.
+
+  Note it writes straight into `bytes.data(buf)`, which does not publish the
+  buffer's length — call `bytes.set_length(buf, n)` before
+  `bytes.to_string(buf, n)` or the conversion returns an empty string with no
+  error. Callers passing the bytes to another extern never need this.
+
 ## [0.509.0]
 
 ### Added

std/io/aether_io.h

@@ -98,6 +98,11 @@ int io_fd_write_n(int fd, const char* data, int length);
 // failure. The string is binary-safe (carries explicit length;
 // embedded NULs survive).
 _tuple_ptrintstr_io io_fd_read_n_tuple(int fd, int n);
+/* Zero-allocation read into a caller-owned buffer (#1471). Returns the byte
+ * count, 0 at EOF, or -1 on error. Returns as soon as ANY bytes are available
+ * (unlike io_fd_read_n_tuple, which loops to fill) so it is usable on a live
+ * pipe. */
+int io_fd_read_into_raw(int fd, void* buf, int length);
 
 // Read one '\n'-delimited line from `fd`. Trailing '\n' is stripped
 // (a trailing '\r' before it is also stripped, so CRLF input yields

std/io/module.ae

@@ -15,11 +15,12 @@ exports(
     io_perror_raw, io_errno_message_raw,
     io_fd_open_read_tuple, io_fd_open_write_tuple, io_fd_close_raw,
     io_fd_write_n, io_fd_read_n_tuple, io_fd_read_line_tuple,
+    io_fd_read_into_raw,
     read_file, write_file, append_file, delete_file, file_info,
     setenv, unsetenv,
     stderr_write, stdout_write, perror, errno_message,
     fd_open_read, fd_open_write, fd_close,
-    fd_write_n, fd_read_n, fd_read_line
+    fd_write_n, fd_read_n, fd_read_into, fd_read_line
 )
 
 // Console I/O — infallible in practice
@@ -78,6 +79,7 @@ extern io_fd_write_n(fd: int, data: string, length: int) -> int
 // was never reclaimed, leaking one buffer per read. aether_heap_str_free
 // dispatches on the magic header (string_release here); never a literal.
 extern io_fd_read_n_tuple(fd: int, n: int) -> (string @heap, int, string)
+extern io_fd_read_into_raw(fd: int, buf: ptr, length: int) -> int
 extern io_fd_read_line_tuple(fd: int) -> (string @heap, string)
 
 // string_concat for owned-copy duplication
@@ -248,6 +250,39 @@ fd_read_n(fd: int, n: int) -> {
     return io_fd_read_n_tuple(fd, n)
 }
 
+// Zero-allocation sibling of `fd_read_n` (#1471): read into a buffer the
+// caller owns instead of minting a fresh string per call. Mirrors
+// `fs.pread_into`, which is the same shape for files.
+//
+// Returns (n, err). `n == 0` means EOF; `0 < n < length` is a normal SHORT
+// READ, not an error — unlike `fd_read_n`, this returns as soon as any bytes
+// are available rather than looping to fill the buffer. That is precisely
+// what makes it usable on a live pipe: waiting to fill would stall until the
+// producer happened to send a whole buffer's worth.
+//
+// Loop until it returns 0 to consume a stream to EOF. `buf` comes from
+// `bytes.new(length)`; pass `bytes.data(buf)` and the same `length` used to
+// allocate it. Reading a frame at a time this way avoids ~250 MB/s of
+// allocator churn at 1080p30 versus the allocating form.
+//
+// GOTCHA: this writes straight into `data()`'s region, which does not update
+// the buffer's own length. Call `bytes.set_length(buf, n)` afterwards before
+// `bytes.to_string(buf, n)`, or you get an empty string back with no error:
+//
+//     n, e = io.fd_read_into(fd, bytes.data(buf), cap)
+//     _ = bytes.set_length(buf, n)          // <- required
+//     s = bytes.to_string(buf, n)
+//
+// A caller passing the bytes straight to another extern (the video case)
+// never needs this — it only matters when converting to a string.
+fd_read_into(fd: int, buf: ptr, length: int) -> int! {
+    n = io_fd_read_into_raw(fd, buf, length)
+    if n < 0 {
+        return 0, "fd_read_into failed"
+    }
+    return n, ""
+}
+
 // Read one '\n'-delimited line from `fd`. The trailing '\n' is
 // stripped (and a '\r' before it for CRLF input). Returns (line, err):
 //   - Normal line: (content, "").

tests/regression/test_fd_read_into.ae

@@ -0,0 +1,114 @@
+// io.fd_read_into — zero-allocation incremental read from a raw fd (#1471).
+//
+// The issue: `std.os.run_pipe` hands back a readable fd, but the only way to
+// consume it was `fd_read_n`, which allocates a fresh string per call. For the
+// streaming-decode case that motivated the ask — one video frame per read, 8 MB
+// at 1080p, 30 times a second — that is ~250 MB/s of allocator churn a reused
+// caller-owned buffer avoids entirely. `fs.pread_into` was already the same
+// shape for files; this is the fd equivalent.
+//
+// What this pins:
+//   1. bytes arrive INCREMENTALLY, before the child exits (the property that
+//      makes streaming possible at all — a drain-to-completion read would
+//      block until EOF and defeat the purpose)
+//   2. ONE buffer serves every read
+//   3. EOF is n == 0 with no error, so `while n > 0` terminates
+//   4. a short read is not an error
+//   5. bad arguments are refused rather than silently misbehaving
+import std.os
+import std.io
+import std.bytes
+import std.string
+import std.list
+
+check(cond: int, label: string) -> int {
+    if cond == 1 {
+        println("  PASS ${label}")
+        return 0
+    }
+    println("  FAIL ${label}")
+    return 1
+}
+
+main() {
+    println("=== io.fd_read_into ===")
+    fails = 0
+
+    // The child writes two chunks with a gap between them, to fd 3 — the
+    // AETHER_IPC_FD channel run_pipe sets up. If reads were drain-to-
+    // completion, the first would not return until the child exited.
+    args = list.new()
+    _ = list.add(args, "-c")
+    _ = list.add(args, "printf 'frame1' >&3; sleep 0.2; printf 'frame2' >&3")
+    fd, pid, err = os.run_pipe("sh", args, null)
+    list.free(args)
+    if err != "" {
+        println("  SKIP: run_pipe unavailable (${err})")
+        string.free(err)
+        return
+    }
+    string.free(err)
+
+    // One buffer, reused for every read. This is the point of the API.
+    cap = 64
+    buf = bytes.new(cap)
+
+    got1 = ""
+    got2 = ""
+    reads = 0
+    total = 0
+    n = 1
+    while n > 0 {
+        n, e = io.fd_read_into(fd, bytes.data(buf), cap)
+        if e != "" {
+            fails = fails + check(0, "read errored: ${e}")
+            n = 0
+        } else {
+            if n > 0 {
+                // Writing through data() does not publish the length; the
+                // buffer still thinks it is empty until set_length says
+                // otherwise. Only matters when converting to a string.
+                _ = bytes.set_length(buf, n)
+                s = bytes.to_string(buf, n)
+                if reads == 0 { got1 = string.concat(s, "") }
+                if reads == 1 { got2 = string.concat(s, "") }
+                string.free(s)
+                reads = reads + 1
+                total = total + n
+            }
+        }
+        string.free(e)
+    }
+
+    fails = fails + check(string.equals(got1, "frame1"), "first chunk arrives before the child exits")
+    fails = fails + check(string.equals(got2, "frame2"), "second chunk arrives after it is written")
+    fails = fails + check(reads == 2, "two incremental reads, not one drained blob (got ${reads})")
+    fails = fails + check(total == 12, "12 bytes total (got ${total})")
+    fails = fails + check(n == 0, "EOF reported as n == 0, so the loop terminates")
+
+    string.free(got1)
+    string.free(got2)
+    bytes.free(buf)
+    _ = io.fd_close(fd)
+    _st, _we = os.wait_pid(pid)
+    string.free(_we)
+
+    // Argument guards: refuse rather than read into nothing.
+    b2 = bytes.new(8)
+    bad_fd, e1 = io.fd_read_into(0 - 1, bytes.data(b2), 8)
+    fails = fails + check(bad_fd == 0 && string.length(e1) > 0, "negative fd is refused")
+    string.free(e1)
+
+    zero_len, e2 = io.fd_read_into(0, bytes.data(b2), 0)
+    fails = fails + check(zero_len == 0 && string.length(e2) > 0, "zero length is refused")
+    string.free(e2)
+    bytes.free(b2)
+
+    println("")
+    if fails == 0 {
+        println("All PASS")
+    } else {
+        println("${fails} FAILURE(S)")
+        exit(1)
+    }
+}