diff --git a/ChangeLog.md b/ChangeLog.md index d1c8711a9870a..ac00d4ee9fcf0 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 0383adcc4f88f..3101fad9b7a7b 100644 --- a/src/lib/libpipefs.js +++ b/src/lib/libpipefs.js @@ -7,8 +7,10 @@ addToLibrary({ $PIPEFS__postset: () => addAtInit('PIPEFS.root = FS.mount(PIPEFS, {}, null);'), $PIPEFS: { - BUCKET_BUFFER_SIZE: 1024 * 8, // 8KiB Buffer 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 +21,13 @@ addToLibrary({ }, createPipe() { var pipe = { - buckets: [], + // 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). Buckets are freed once both counts + // report POLLHUP (Linux semantics). The buffer is freed once both counts // reach 0. writerCount: 1, writeClosed: false, @@ -32,12 +38,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); @@ -74,9 +74,29 @@ addToLibrary({ }; }, queuedBytes(pipe) { - var n = 0; - for (var bucket of pipe.buckets) n += bucket.offset - bucket.roffset; - return n; + 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) { +#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; + }, + // 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) { @@ -154,8 +174,6 @@ addToLibrary({ assert(buffer instanceof ArrayBuffer || ArrayBuffer.isView(buffer)); #endif #endif - var data = buffer.subarray(offset, offset + length); - if (length <= 0) { return 0; } @@ -165,49 +183,18 @@ 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++; - } - } - - 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); + 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 // 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; @@ -229,71 +216,29 @@ 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 room = PIPEFS.CAPACITY - PIPEFS.queuedBytes(pipe); + if (room <= 0 || (dataLen <= {{{ cDefs.PIPE_BUF }}} && dataLen > room)) { + throw new FS.ErrnoError({{{ cDefs.EAGAIN }}}); } - - 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 (dataLen > room) { + data = data.subarray(0, room); + dataLen = room; } - -#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); + if (pipe.count + dataLen > pipe.buffer.length) { + PIPEFS.grow(pipe, pipe.count + dataLen); } - - 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 (remElements > 0) { - var newBucket = { - buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE), - offset: data.byteLength, - roffset: 0 - }; - pipe.buckets.push(newBucket); - newBucket.buffer.set(data); + // 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.count += dataLen; pipe.readNode.notifyListeners({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}}); return dataLen; @@ -314,7 +259,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..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": 272928, - "a.out.nodebug.wasm": 589443, - "total": 862371, + "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 5fa6b3571ebfb..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); @@ -63,13 +87,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); @@ -81,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; }