diff --git a/CHANGELOG.md b/CHANGELOG.md index 6de1c27cc..dfb264cb0 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/std/io/aether_io.c b/std/io/aether_io.c index 72ae5da58..8d67adc2b 100644 --- a/std/io/aether_io.c +++ b/std/io/aether_io.c @@ -41,6 +41,7 @@ _tuple_int_string_io io_fd_open_write_tuple(const char* p) { (void)p; _tuple_int const char* io_fd_close_raw(int fd) { (void)fd; return "filesystem disabled"; } int io_fd_write_n(int fd, const char* d, int n) { (void)fd; (void)d; (void)n; return -1; } _tuple_ptrintstr_io io_fd_read_n_tuple(int fd, int n) { (void)fd; (void)n; _tuple_ptrintstr_io t; t._0 = NULL; t._1 = 0; t._2 = "filesystem disabled"; return t; } +int io_fd_read_into_raw(int fd, void* buf, int length) { (void)fd; (void)buf; (void)length; return -1; } _tuple_ptrstr_io io_fd_read_line_tuple(int fd) { (void)fd; _tuple_ptrstr_io t; t._0 = NULL; t._1 = "filesystem disabled"; return t; } #else @@ -431,6 +432,41 @@ _tuple_ptrintstr_io io_fd_read_n_tuple(int fd, int n) { return out; } +/* Zero-allocation sibling of io_fd_read_n_tuple: read into a CALLER-owned + * buffer instead of minting a fresh AetherString per call. + * + * Why it exists (#1471): the streaming-decode case reads one video frame per + * iteration. At 1080p a frame is ~8 MB, so allocating a fresh string per frame + * at 30fps is ~250 MB/s of pure allocator churn that a reused buffer avoids + * entirely. `fs.pread_into` is the same shape for files; this is the fd + * equivalent, so a caller streaming from a pipe gets the same option as one + * reading a file. + * + * Returns the byte count, or -1 on error. Like read(2) and unlike + * io_fd_read_n_tuple, this returns as soon as ANY bytes are available rather + * than looping to fill the buffer — that is what makes it usable on a live + * pipe, where waiting to fill would stall until the producer sent a full + * buffer's worth. n == 0 means EOF; 0 < n < length is a normal short read, not + * an error. EINTR is retried. + * + * The buffer is caller-owned (std.bytes) and its capacity is trusted the same + * way fs_pread_into_raw trusts its own — the Aether wrapper is the guard. */ +int io_fd_read_into_raw(int fd, void* buf, int length) { + if (fd < 0 || !buf || length <= 0) return -1; + for (;;) { +#ifdef _WIN32 + int r = AE_FD_READ(fd, (char*)buf, length); +#else + long r = AE_FD_READ(fd, (char*)buf, length); +#endif + if (r >= 0) return (int)r; /* 0 == EOF */ +#ifndef _WIN32 + if (errno == EINTR) continue; +#endif + return -1; + } +} + _tuple_ptrstr_io io_fd_read_line_tuple(int fd) { _tuple_ptrstr_io out; out._0 = (void*)string_empty(); diff --git a/std/io/aether_io.h b/std/io/aether_io.h index b7c4e5702..cabc85d26 100644 --- a/std/io/aether_io.h +++ b/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 diff --git a/std/io/module.ae b/std/io/module.ae index 6889131af..a3b472768 100644 --- a/std/io/module.ae +++ b/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, ""). diff --git a/tests/regression/test_fd_read_into.ae b/tests/regression/test_fd_read_into.ae new file mode 100644 index 000000000..3408993e4 --- /dev/null +++ b/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) + } +}