Repository navigation
[FS] Grow pipe buffers on demand as a ring - #28098
PedroRosalba wants to merge 5 commits into
Conversation
Since `PIPEFS` never blocks on `read()` or `write()` (already treating `read()` as non-blocking regardless of `O_NONBLOCK`), apply the 64 KiB pipe capacity limit to all writes in `libpipefs.js`. With pipes bounded at 64 KiB in all cases, replace the multi-bucket list in `libpipefs.js` with a single 64 KiB `Uint8Array` buffer that resets its cursors when fully drained and lazily compacts unread bytes with `copyWithin()` only when a write would run past the end of the buffer. See: emscripten-core#27831
A pipe allocated its full 64 KiB buffer up front. Programs that create many mostly-empty pipes pay for all of it: Rust's mio creates one waker pipe per Poll, so one per tokio runtime, and each holds a byte or two. Start each pipe at 8 KiB and double up to the 64 KiB capacity as data arrives, like Linux, which allocates pipe pages as they are written. Store the data as a ring, so writes near a full pipe wrap around instead of compacting with copyWithin. Growing unwraps into the new buffer, at most three times per pipe. 1000 pipes holding one byte each now use 8,000 KiB of buffers instead of 64,000 KiB. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the pipe implementation in libpipefs.js from a bucket-based system to a dynamic ring buffer that starts small and doubles in size as needed, optimizing memory usage for mostly-empty pipes. It also updates the changelog and adds robust test coverage for buffer wrap-around and growth behaviors. The feedback suggests adding safety assertions in the new grow and copyOut helper functions to enforce capacity limits and prevent potential out-of-bounds writes.
| grow(pipe, needed) { | ||
| var size = pipe.buffer.length; | ||
| while (size < needed) size *= 2; | ||
| var buffer = new Uint8Array(size); | ||
| PIPEFS.copyOut(pipe, buffer, 0, pipe.count); | ||
| pipe.buffer = buffer; | ||
| pipe.head = 0; | ||
| }, |
There was a problem hiding this comment.
To prevent potential buffer overflows or unexpected growth beyond the maximum allowed capacity (PIPEFS.CAPACITY), we should add an assertion at the beginning of the grow function. This ensures that any future changes or unexpected callers of grow do not violate the capacity constraints.
grow(pipe, needed) {
#if ASSERTIONS
assert(needed <= PIPEFS.CAPACITY);
#endif
var size = pipe.buffer.length;
while (size < needed) size *= 2;
var buffer = new Uint8Array(size);
PIPEFS.copyOut(pipe, buffer, 0, pipe.count);
pipe.buffer = buffer;
pipe.head = 0;
},There was a problem hiding this comment.
Done in bf37a3d.
AI-generated with Claude, reviewed by me before posting.
| copyOut(pipe, dest, offset, length) { | ||
| var first = Math.min(length, pipe.buffer.length - pipe.head); | ||
| dest.set(pipe.buffer.subarray(pipe.head, pipe.head + first), offset); | ||
| if (first < length) { | ||
| dest.set(pipe.buffer.subarray(0, length - first), offset + first); | ||
| } | ||
| }, |
There was a problem hiding this comment.
To ensure safety and prevent potential out-of-bounds writes on the destination buffer, we should add an assertion in copyOut to verify that the destination buffer has sufficient capacity to hold the requested length of bytes starting at offset.
copyOut(pipe, dest, offset, length) {
#if ASSERTIONS
assert(offset + length <= dest.length);
#endif
var first = Math.min(length, pipe.buffer.length - pipe.head);
dest.set(pipe.buffer.subarray(pipe.head, pipe.head + first), offset);
if (first < length) {
dest.set(pipe.buffer.subarray(0, length - first), offset + first);
}
},There was a problem hiding this comment.
TypedArray.set already throws a RangeError when the source doesn't fit at offset, in every build, so an overflow here already fails loudly. The assertion wouldn't catch anything new.
AI-generated with Claude, reviewed by me before posting.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Follow-up to #28087, as discussed in #28087 (comment). This is stacked on #28087, so only the last commit is new here. I'll rebase once #28087 lands.
#28087 gives each pipe a single 64 KiB buffer, allocated when the pipe is created. Linux allocates pipe pages as data is written, so an empty pipe holds no data pages. The difference adds up when a program has many mostly-empty pipes. Rust's
miocreates one waker pipe perPoll, which means one per tokio runtime, and each of those pipes holds a byte or two.This change:
PIPEFS.CAPACITY, when a write needs more room.head,count). A write near a full pipe wraps around instead of compacting withcopyWithin. Growing copies the queued bytes, unwrapped, into the new buffer, at most three times per pipe.PIPE_BUFrule, poll readiness,EPIPEand EOF.Not done: buffers don't shrink. A pipe that once held 64 KiB keeps that buffer until it is closed.
WasmFS is unchanged. Its pipes keep data in a
std::queue<uint8_t>, which already grows with its contents.Tests:
test_pipe_capacityhas a new section that sends bytes as a running counter, so data out of order fails an assert. The section wraps inside the initial buffer, grows while the data is wrapped, wraps the full-size buffer, and checks that a read doesn't write before its destination. Each of these breaks the test: removing the wrap inwrite(), removing the modulo onheadinread(), and copying without unwrapping when growing. With 1000 pipes holding one byte each, the buffers total 65,536,000 bytes before this change and 8,192,000 after.AI-generated with Claude, reviewed by me before posting.