From 0bae681b014c433a1951931fd051d9ce8e6ab445 Mon Sep 17 00:00:00 2001 From: Sam Clegg Date: Tue, 27 Oct 2026 14:48:18 -0700 Subject: [PATCH 1/3] [FS] Limit pipes to 64 KiB in all cases and use a single buffer 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: #27831 --- ChangeLog.md | 4 +- src/lib/libpipefs.js | 140 ++++-------------- .../test_codesize_hello_dylink_all.json | 4 +- test/core/test_pipe_capacity.c | 7 - 4 files changed, 33 insertions(+), 122 deletions(-) diff --git a/ChangeLog.md b/ChangeLog.md index e3bb64fecb6d5..88051b83aa750 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -20,8 +20,8 @@ See docs/process.md for more on how version tagging works. 6.0.14 (in development) ----------------------- -- Pipes now have a 64 KiB capacity: non-blocking writes to a full pipe fail - with `EAGAIN`. For now, blocking writes remain unbounded. (#27831) +- Pipes now have a 64 KiB capacity: writes to a full pipe fail with `EAGAIN`. + (#27831) - When compiling and linking in a single command, auxiliary compiler output files (such as `.dwo` files from `-gsplit-dwarf` or `.json` files from `-ftime-trace`) are now named `-.`, matching the behavior diff --git a/src/lib/libpipefs.js b/src/lib/libpipefs.js index 2c5d0d152c6cd..ba9f119038fbf 100644 --- a/src/lib/libpipefs.js +++ b/src/lib/libpipefs.js @@ -8,7 +8,6 @@ addToLibrary({ $PIPEFS__postset: () => addAtInit('PIPEFS.root = FS.mount(PIPEFS, {}, null);'), $PIPEFS__deps: ['$FS'], $PIPEFS: { - BUCKET_BUFFER_SIZE: 1024 * 8, // 8KiB Buffer CAPACITY: 1024 * 64, // as on Linux // Queued bytes above which a PIPE_BUF write no longer fits. THRESHOLD: 1024 * 64 - {{{ cDefs.PIPE_BUF }}}, @@ -20,9 +19,11 @@ addToLibrary({ }, createPipe() { var pipe = { - buckets: [], + buffer: new Uint8Array(PIPEFS.CAPACITY), + offset: 0, + roffset: 0, // Open write ends. When it drops to 0 the reader sees EOF and poll must - // report POLLHUP (Linux semantics). Buckets are freed once both counts + // report POLLHUP (Linux semantics). The buffer is freed once both counts // reach 0. writerCount: 1, writeClosed: false, @@ -33,12 +34,6 @@ addToLibrary({ timestamp: new Date(), }; - pipe.buckets.push({ - buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE), - offset: 0, - roffset: 0 - }); - var rName = PIPEFS.nextname(); var wName = PIPEFS.nextname(); var rNode = FS.createNode(PIPEFS.root, rName, {{{ cDefs.S_IFIFO }}}, 0); @@ -75,9 +70,7 @@ addToLibrary({ }; }, queuedBytes(pipe) { - var n = 0; - for (var bucket of pipe.buckets) n += bucket.offset - bucket.roffset; - return n; + return pipe.offset - pipe.roffset; }, stream_ops: { getattr(stream) { @@ -155,8 +148,6 @@ addToLibrary({ assert(buffer instanceof ArrayBuffer || ArrayBuffer.isView(buffer)); #endif #endif - var data = buffer.subarray(offset, offset + length); - if (length <= 0) { return 0; } @@ -166,49 +157,21 @@ addToLibrary({ throw new FS.ErrnoError({{{ cDefs.EAGAIN }}}); } var toRead = Math.min(currentLength, length); - - var totalRead = toRead; - var toRemove = 0; - - for (var bucket of pipe.buckets) { - var bucketSize = bucket.offset - bucket.roffset; - - if (toRead <= bucketSize) { - var tmpSlice = bucket.buffer.subarray(bucket.roffset, bucket.offset); - if (toRead < bucketSize) { - tmpSlice = tmpSlice.subarray(0, toRead); - bucket.roffset += toRead; - } else { - toRemove++; - } - data.set(tmpSlice); - break; - } else { - var tmpSlice = bucket.buffer.subarray(bucket.roffset, bucket.offset); - data.set(tmpSlice); - data = data.subarray(tmpSlice.byteLength); - toRead -= tmpSlice.byteLength; - toRemove++; - } + buffer.set(pipe.buffer.subarray(pipe.roffset, pipe.roffset + toRead), offset); + pipe.roffset += toRead; + // Reset cursors when fully drained so subsequent writes start at 0 + // without needing to shift data. + if (pipe.roffset == pipe.offset) { + pipe.roffset = pipe.offset = 0; } - if (toRemove && toRemove == pipe.buckets.length) { - // Do not generate excessive garbage in use cases such as - // write several bytes, read everything, write several bytes, read everything... - toRemove--; - pipe.buckets[toRemove].offset = 0; - pipe.buckets[toRemove].roffset = 0; - } - - pipe.buckets.splice(0, toRemove); - // Wake POLLOUT waiters on the transition to writable: listeners only // notify on readiness changes, and a waiter only exists while the // pipe is above the threshold. - if (currentLength > PIPEFS.THRESHOLD && currentLength - totalRead <= PIPEFS.THRESHOLD) { + if (currentLength > PIPEFS.THRESHOLD && currentLength - toRead <= PIPEFS.THRESHOLD) { pipe.writeNode.notifyListeners({{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}}); } - return totalRead; + return toRead; }, write(stream, buffer, offset, length, position /* ignored */) { var pipe = stream.node.pipe; @@ -230,71 +193,26 @@ addToLibrary({ throw new FS.ErrnoError({{{ cDefs.EPIPE }}}); } + // Behave as if the write end is always non-blocking. // PIPE_BUF rule: a small write is all-or-nothing, a larger one takes // what fits. - if (stream.flags & {{{ cDefs.O_NONBLOCK }}}) { - var room = PIPEFS.CAPACITY - PIPEFS.queuedBytes(pipe); - if (room <= 0 || (dataLen <= {{{ cDefs.PIPE_BUF }}} && dataLen > room)) { - throw new FS.ErrnoError({{{ cDefs.EAGAIN }}}); - } - if (dataLen > room) { - data = data.subarray(0, room); - dataLen = room; - } - } - - var currBucket = null; - - if (pipe.buckets.length == 0) { - currBucket = { - buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE), - offset: 0, - roffset: 0 - }; - pipe.buckets.push(currBucket); - } else { - currBucket = pipe.buckets[pipe.buckets.length - 1]; - } - -#if ASSERTIONS - assert(currBucket.offset <= PIPEFS.BUCKET_BUFFER_SIZE); -#endif - - var freeBytesInCurrBuffer = PIPEFS.BUCKET_BUFFER_SIZE - currBucket.offset; - if (freeBytesInCurrBuffer >= dataLen) { - currBucket.buffer.set(data, currBucket.offset); - currBucket.offset += dataLen; - pipe.readNode.notifyListeners({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}}); - return dataLen; - } else if (freeBytesInCurrBuffer > 0) { - currBucket.buffer.set(data.subarray(0, freeBytesInCurrBuffer), currBucket.offset); - currBucket.offset += freeBytesInCurrBuffer; - data = data.subarray(freeBytesInCurrBuffer, data.byteLength); + var room = PIPEFS.CAPACITY - PIPEFS.queuedBytes(pipe); + if (room <= 0 || (dataLen <= {{{ cDefs.PIPE_BUF }}} && dataLen > room)) { + throw new FS.ErrnoError({{{ cDefs.EAGAIN }}}); } - - var numBuckets = (data.byteLength / PIPEFS.BUCKET_BUFFER_SIZE) | 0; - var remElements = data.byteLength % PIPEFS.BUCKET_BUFFER_SIZE; - - for (var i = 0; i < numBuckets; i++) { - var newBucket = { - buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE), - offset: PIPEFS.BUCKET_BUFFER_SIZE, - roffset: 0 - }; - pipe.buckets.push(newBucket); - newBucket.buffer.set(data.subarray(0, PIPEFS.BUCKET_BUFFER_SIZE)); - data = data.subarray(PIPEFS.BUCKET_BUFFER_SIZE, data.byteLength); + if (dataLen > room) { + data = data.subarray(0, room); + dataLen = room; } - - if (remElements > 0) { - var newBucket = { - buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE), - offset: data.byteLength, - roffset: 0 - }; - pipe.buckets.push(newBucket); - newBucket.buffer.set(data); + // Lazily compact unread bytes to the start of the buffer only when a + // write would run past the end of the 64 KiB array. + if (pipe.offset + dataLen > PIPEFS.CAPACITY) { + pipe.buffer.copyWithin(0, pipe.roffset, pipe.offset); + pipe.offset -= pipe.roffset; + pipe.roffset = 0; } + pipe.buffer.set(data, pipe.offset); + pipe.offset += dataLen; pipe.readNode.notifyListeners({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}}); return dataLen; @@ -315,7 +233,7 @@ addToLibrary({ pipe.writeNode.notifyListeners({{{ cDefs.POLLERR }}} | {{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}}); } if (!pipe.readerCount && !pipe.writerCount) { - pipe.buckets = null; + pipe.buffer = null; } } }, diff --git a/test/codesize/test_codesize_hello_dylink_all.json b/test/codesize/test_codesize_hello_dylink_all.json index 46a391f5d3793..93c50cca7de8c 100644 --- a/test/codesize/test_codesize_hello_dylink_all.json +++ b/test/codesize/test_codesize_hello_dylink_all.json @@ -1,7 +1,7 @@ { - "a.out.js": 272928, + "a.out.js": 272151, "a.out.nodebug.wasm": 589443, - "total": 862371, + "total": 861594, "sent": [ "IMG_Init", "IMG_Load", diff --git a/test/core/test_pipe_capacity.c b/test/core/test_pipe_capacity.c index 5fa6b3571ebfb..86f82bfbef839 100644 --- a/test/core/test_pipe_capacity.c +++ b/test/core/test_pipe_capacity.c @@ -63,13 +63,6 @@ int main() { assert(drain(p[0]) == CAPACITY); assert(writable(p[1])); - // A blocking writer is not held to the capacity. - assert(fcntl(p[1], F_SETFL, 0) == 0); - for (int i = 0; i < 256; i++) - assert(write(p[1], chunk, sizeof chunk) == sizeof chunk); - assert(drain(p[0]) == 1 << 20); - assert(fcntl(p[1], F_SETFL, O_NONBLOCK) == 0); - assert(write(p[1], "ab", 2) == 2); assert(close(p[1]) == 0); assert(read(p[0], buf, sizeof buf) == 2); From 0e03fbbd1f0b83a7c72703a057518897be2c188e Mon Sep 17 00:00:00 2001 From: PedroRosalba Date: Wed, 28 Oct 2026 23:27:33 +0000 Subject: [PATCH 2/3] [FS] Grow pipe buffers on demand as a ring 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 --- src/lib/libpipefs.js | 62 +++++++++++++------ .../test_codesize_hello_dylink_all.json | 6 +- test/core/test_pipe_capacity.c | 45 ++++++++++++++ 3 files changed, 91 insertions(+), 22 deletions(-) diff --git a/src/lib/libpipefs.js b/src/lib/libpipefs.js index ba9f119038fbf..fdf11a7f73ee0 100644 --- a/src/lib/libpipefs.js +++ b/src/lib/libpipefs.js @@ -9,6 +9,9 @@ addToLibrary({ $PIPEFS__deps: ['$FS'], $PIPEFS: { CAPACITY: 1024 * 64, // as on Linux + // A new pipe starts with a small buffer that doubles as data arrives, so + // the many mostly-empty pipes some programs create stay cheap. + INITIAL_SIZE: 1024 * 8, // Queued bytes above which a PIPE_BUF write no longer fits. THRESHOLD: 1024 * 64 - {{{ cDefs.PIPE_BUF }}}, @@ -19,9 +22,11 @@ addToLibrary({ }, createPipe() { var pipe = { - buffer: new Uint8Array(PIPEFS.CAPACITY), - offset: 0, - roffset: 0, + // Ring buffer: `count` unread bytes starting at `head`, wrapping + // around the end of `buffer`. + buffer: new Uint8Array(PIPEFS.INITIAL_SIZE), + head: 0, + count: 0, // Open write ends. When it drops to 0 the reader sees EOF and poll must // report POLLHUP (Linux semantics). The buffer is freed once both counts // reach 0. @@ -70,7 +75,26 @@ addToLibrary({ }; }, queuedBytes(pipe) { - return pipe.offset - pipe.roffset; + return pipe.count; + }, + // Grows the buffer (doubling, up to CAPACITY) until `needed` bytes fit, + // unwrapping the queued bytes to the start of the new buffer. + 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; + }, + // Copies `length` queued bytes starting at `head` into `dest` at `offset`, + // in at most two pieces when the data wraps. + 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); + } }, stream_ops: { getattr(stream) { @@ -157,13 +181,10 @@ addToLibrary({ throw new FS.ErrnoError({{{ cDefs.EAGAIN }}}); } var toRead = Math.min(currentLength, length); - buffer.set(pipe.buffer.subarray(pipe.roffset, pipe.roffset + toRead), offset); - pipe.roffset += toRead; - // Reset cursors when fully drained so subsequent writes start at 0 - // without needing to shift data. - if (pipe.roffset == pipe.offset) { - pipe.roffset = pipe.offset = 0; - } + PIPEFS.copyOut(pipe, buffer, offset, toRead); + pipe.count -= toRead; + // Restart at 0 once drained so the next write does not wrap. + pipe.head = pipe.count ? (pipe.head + toRead) % pipe.buffer.length : 0; // Wake POLLOUT waiters on the transition to writable: listeners only // notify on readiness changes, and a waiter only exists while the @@ -204,15 +225,18 @@ addToLibrary({ data = data.subarray(0, room); dataLen = room; } - // Lazily compact unread bytes to the start of the buffer only when a - // write would run past the end of the 64 KiB array. - if (pipe.offset + dataLen > PIPEFS.CAPACITY) { - pipe.buffer.copyWithin(0, pipe.roffset, pipe.offset); - pipe.offset -= pipe.roffset; - pipe.roffset = 0; + if (pipe.count + dataLen > pipe.buffer.length) { + PIPEFS.grow(pipe, pipe.count + dataLen); + } + // Write after the queued bytes, wrapping around the end of the buffer. + var size = pipe.buffer.length; + var tail = (pipe.head + pipe.count) % size; + var first = Math.min(dataLen, size - tail); + pipe.buffer.set(data.subarray(0, first), tail); + if (first < dataLen) { + pipe.buffer.set(data.subarray(first), 0); } - pipe.buffer.set(data, pipe.offset); - pipe.offset += dataLen; + pipe.count += dataLen; pipe.readNode.notifyListeners({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}}); return dataLen; diff --git a/test/codesize/test_codesize_hello_dylink_all.json b/test/codesize/test_codesize_hello_dylink_all.json index 93c50cca7de8c..5d864421f397b 100644 --- a/test/codesize/test_codesize_hello_dylink_all.json +++ b/test/codesize/test_codesize_hello_dylink_all.json @@ -1,7 +1,7 @@ { - "a.out.js": 272151, - "a.out.nodebug.wasm": 589443, - "total": 861594, + "a.out.js": 272442, + "a.out.nodebug.wasm": 588169, + "total": 860611, "sent": [ "IMG_Init", "IMG_Load", diff --git a/test/core/test_pipe_capacity.c b/test/core/test_pipe_capacity.c index 86f82bfbef839..57ac9178aea16 100644 --- a/test/core/test_pipe_capacity.c +++ b/test/core/test_pipe_capacity.c @@ -31,6 +31,30 @@ size_t drain(int fd) { return total; } +// Bytes go through the pipe as a running counter so wrap-around and growth +// bugs show up as data out of order. +unsigned char next_out, next_in; + +void write_seq(int fd, size_t n) { + static unsigned char data[CAPACITY]; + for (size_t i = 0; i < n; i++) + data[i] = next_out++; + assert(write(fd, data, n) == n); +} + +void read_seq(int fd, size_t n) { + // The guard catches a read that writes before the start of `data`. + static struct { + unsigned char guard[CAPACITY]; + unsigned char data[CAPACITY]; + } dest; + assert(read(fd, dest.data, n) == n); + for (size_t i = 0; i < n; i++) + assert(dest.data[i] == next_in++); + for (size_t i = 0; i < CAPACITY; i++) + assert(dest.guard[i] == 0); +} + int main() { int p[2]; assert(pipe(p) == 0); @@ -74,6 +98,27 @@ int main() { assert(write(p[1], "x", 1) == -1 && errno == EPIPE); assert(close(p[1]) == 0); + // Wrap around the end of the initial buffer, grow while wrapped, then wrap + // the full-size buffer. + assert(pipe(p) == 0); + assert(fcntl(p[1], F_SETFL, O_NONBLOCK) == 0); + write_seq(p[1], 6000); + read_seq(p[0], 5000); + write_seq(p[1], 4000); + write_seq(p[1], 8000); + read_seq(p[0], 2000); + write_seq(p[1], 50000); + read_seq(p[0], 60000); + write_seq(p[1], 40000); + read_seq(p[0], 6000); + read_seq(p[0], 35000); + assert(drain(p[0]) == 0); + write_seq(p[1], CAPACITY); + assert(!writable(p[1])); + read_seq(p[0], CAPACITY); + assert(close(p[0]) == 0); + assert(close(p[1]) == 0); + printf("done\n"); return 0; } From bf37a3dd5e1c99440b6f7920ea2cbf36d505e520 Mon Sep 17 00:00:00 2001 From: PedroRosalba Date: Thu, 29 Oct 2026 10:39:29 +0000 Subject: [PATCH 3/3] pipefs: assert grow stays within the pipe capacity Co-Authored-By: Claude Opus 5.5 --- src/lib/libpipefs.js | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/lib/libpipefs.js b/src/lib/libpipefs.js index fdf11a7f73ee0..ec6b3181dd1d6 100644 --- a/src/lib/libpipefs.js +++ b/src/lib/libpipefs.js @@ -80,6 +80,9 @@ addToLibrary({ // Grows the buffer (doubling, up to CAPACITY) until `needed` bytes fit, // unwrapping the queued bytes to the start of the new buffer. 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);