Skip to content

[FS] Limit pipes to 64 KiB in all cases and use a single buffer #28087

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
sbc100 wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

[FS] Limit pipes to 64 KiB in all cases and use a single buffer #28087

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ChangeLog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 `<output>-<input>.<ext>`, matching the behavior
Expand Down
140 changes: 29 additions & 111 deletions src/lib/libpipefs.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }}},
Expand All @@ -20,9 +19,11 @@ addToLibrary({
},
createPipe() {
var pipe = {
buckets: [],
buffer: new Uint8Array(PIPEFS.CAPACITY),
Comment thread
guybedford marked this conversation as resolved.
Resolved
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,
Expand All @@ -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);
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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;
}
Expand All @@ -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;
Expand All @@ -230,71 +193,26 @@ addToLibrary({
throw new FS.ErrnoError({{{ cDefs.EPIPE }}});
}

// Behave as if the write end is always non-blocking.

@guybedford guybedford Oct 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the behaviour change for blocking writers described in the main comment. If it stays, the ChangeLog entry should say that blocking writes now behave as non-blocking, since the previous entry promised the opposite.

// 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);

@guybedford guybedford Oct 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Near a full pipe every write compacts up to 64 KiB. 100k 512-byte read/write ping-pongs measured 83 ms here versus 47 ms on main. Not a blocker, a two-segment ring (read/write split at the wrap) would avoid it at similar code size.

@sbc100 sbc100 Oct 27, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm.. good point doing a lot of copyWithin is probably fairly costly.

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;
Expand All @@ -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;
}
}
},
Expand Down
4 changes: 2 additions & 2 deletions test/codesize/test_codesize_hello_dylink_all.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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",
Expand Down
7 changes: 0 additions & 7 deletions test/core/test_pipe_capacity.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading