From 0bae681b014c433a1951931fd051d9ce8e6ab445 Mon Sep 17 00:00:00 2001 From: Sam Clegg Date: Tue, 27 Oct 2026 14:48:18 -0700 Subject: [PATCH] [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);