Visitar URL original
perf: stop splitting decompressor input into 16 KiB writes (#30) · gmaclennan/zip-reader@b654075 · GitHub
Skip to content

Commit b654075

Browse files
gmaclennangmaclennan
authored
perf: stop splitting decompressor input into 16 KiB writes (#30)
* perf: stop splitting decompressor input into 16 KiB writes The split capped WebKit's output chunks, since WebKit inflates a whole input write into one chunk, at about 16 MiB instead of 64 MiB. That only matters for near-1032:1 data such as zip bombs or long runs of zeros, and the size check already rejects an oversized chunk before it reaches the caller. Meanwhile the extra writes slow decompression of incompressible data by up to 2x in Node, Chromium, Firefox and WebKit, and large realistic entries by about 5-10% end to end. Reads are already at most 64 KiB, the queue's high-water mark, so that remains the bound on each write. * test: check entry data is read in bounded slices instead of output chunk size * fix: stop Node 22 reading whole deflated entries into memory Node before 24.19 counts each chunk written to DecompressionStream as size 1 against a 16 KiB high-water mark, so pipeThrough read an entry's entire compressed data ahead of the reader (nodejs/node#62986). A 64 MiB entry held 64 MiB after reading one chunk. Feed the decompressor one awaited write at a time instead of using pipeThrough, which bounds read-ahead to a few chunks on every runtime. The decompress-node contract test now checks that an abort reason reaches the reader as the same object, which the upstream error mapping relies on.
1 parent d62e0a6 commit b654075

3 files changed

Lines changed: 71 additions & 36 deletions

File tree

‎src/entry.ts‎

Lines changed: 29 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,6 @@ import { readFields, dosDateTimeToDate } from "./utils.js";
1515
import { LFH_FIELDS } from "./zip-records.js";
1616
import { DecompressionFailed } from "./errors.js";
1717

18-
// WebKit inflates a whole input chunk into one output chunk, so a 64 KiB write
19-
// can emit 64 MiB at once; 16 KiB writes cap that at ~16 MiB
20-
const MAX_DECOMPRESSOR_WRITE = 16384;
21-
2218
const QUEUING_STRATEGY = new ByteLengthQueuingStrategy({
2319
highWaterMark: 65536,
2420
});
@@ -220,13 +216,7 @@ export class ZipEntry {
220216
chunkSize,
221217
);
222218
bytesRead += chunk.byteLength;
223-
if (decompress && chunk.byteLength > MAX_DECOMPRESSOR_WRITE) {
224-
for (let i = 0; i < chunk.byteLength; i += MAX_DECOMPRESSOR_WRITE) {
225-
controller.enqueue(chunk.subarray(i, i + MAX_DECOMPRESSOR_WRITE));
226-
}
227-
} else {
228-
controller.enqueue(chunk);
229-
}
219+
controller.enqueue(chunk);
230220

231221
if (bytesRead >= compressedSize) {
232222
controller.close();
@@ -236,7 +226,8 @@ export class ZipEntry {
236226
let stream: ReadableStream<Uint8Array<ArrayBuffer>> = rawStream;
237227

238228
if (decompress) {
239-
stream = stream.pipeThrough(
229+
stream = pipeThroughOneWriteAtATime(
230+
stream,
240231
new DecompressionStream("deflate-raw") as TransformStream<
241232
Uint8Array<ArrayBuffer>,
242233
Uint8Array<ArrayBuffer>
@@ -356,6 +347,32 @@ function createOutputStream(
356347
);
357348
}
358349

350+
/**
351+
* Like `input.pipeThrough(transform)`, but waits for each write to finish
352+
* before reading the next chunk. Node before 24.19 miscounts backpressure on
353+
* DecompressionStream's input, so pipeThrough reads the whole entry into memory
354+
* ahead of the reader: https://github.com/nodejs/node/pull/62986
355+
*/
356+
function pipeThroughOneWriteAtATime<I, O>(
357+
input: ReadableStream<I>,
358+
transform: TransformStream<I, O>,
359+
): ReadableStream<O> {
360+
const reader = input.getReader();
361+
const writer = transform.writable.getWriter();
362+
(async () => {
363+
for (;;) {
364+
const { done, value } = await reader.read();
365+
if (done) return writer.close();
366+
await writer.write(value);
367+
}
368+
})().catch((error: unknown) => {
369+
// An upstream error reaches the reader as the same object, as with pipeThrough
370+
writer.abort(error).catch(() => {});
371+
reader.cancel(error).catch(() => {});
372+
});
373+
return transform.readable;
374+
}
375+
359376
function describeError(error: unknown): string {
360377
if (error instanceof Error && error.message) return error.message;
361378
const code = (error as { code?: unknown } | null)?.code;

‎test/decompress-node.test.ts‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -52,14 +52,14 @@ describe("Node deflate-raw DecompressionStream", () => {
5252
});
5353

5454
// ZipEntry relies on this to tell upstream errors from decompression errors
55-
it("passes an upstream error through as the same object", async () => {
55+
it("passes an abort reason through as the same object", async () => {
5656
const upstream = new Error("source failed");
57-
const input = new ReadableStream<Uint8Array<ArrayBuffer>>({
58-
pull(controller) {
59-
controller.error(upstream);
60-
},
61-
});
62-
const reader = input.pipeThrough(createInflate()).getReader();
57+
const inflate = createInflate();
58+
const reader = inflate.readable.getReader();
59+
inflate.writable
60+
.getWriter()
61+
.abort(upstream)
62+
.catch(() => {});
6363
await expect(reader.read()).rejects.toBe(upstream);
6464
});
6565
});

‎test/edge-cases.test.ts‎

Lines changed: 35 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1080,24 +1080,42 @@ describe("Edge cases and malformed ZIP handling", () => {
10801080
expect(delivered).toBeLessThanOrEqual(declared);
10811081
}, 30_000);
10821082

1083-
it("bounds the size of a single decompressed chunk", async () => {
1084-
const total = 64 << 20;
1085-
const compressed = await deflateRawZeros(total);
1086-
const zip = buildDeflatedZip("zeros.bin", compressed, total);
1087-
const entry = await firstEntry(new BufferSource(zip), {
1088-
skipCrc32: true,
1089-
});
1090-
const reader = entry.readable().getReader();
1091-
let largestChunk = 0;
1092-
for (let i = 0; i < 8; i++) {
1093-
const { done, value } = await reader.read();
1094-
if (done) break;
1095-
largestChunk = Math.max(largestChunk, value.byteLength);
1083+
it("reads entry data in bounded slices, only as the reader pulls", async () => {
1084+
const data = new Uint8Array(1 << 20);
1085+
for (let i = 0; i < data.length; i += 65536) {
1086+
crypto.getRandomValues(data.subarray(i, i + 65536));
10961087
}
1097-
await reader.cancel();
1098-
// At most 16 KiB written at a time, at deflate's ~1032:1 maximum ratio
1099-
expect(largestChunk).toBeLessThanOrEqual(1032 * 16384);
1100-
}, 30_000);
1088+
const compressed = new Uint8Array(
1089+
await new Response(
1090+
new Blob([data])
1091+
.stream()
1092+
.pipeThrough(new CompressionStream("deflate-raw")),
1093+
).arrayBuffer(),
1094+
);
1095+
const inner = new BufferSource(
1096+
buildDeflatedZip("random.bin", compressed, data.length),
1097+
);
1098+
const reads: number[] = [];
1099+
const source: RandomAccessSource = {
1100+
size: inner.size,
1101+
read(offset, length) {
1102+
reads.push(length);
1103+
return inner.read(offset, length);
1104+
},
1105+
};
1106+
const entry = await firstEntry(source, { skipCrc32: true });
1107+
reads.length = 0;
1108+
1109+
const reader = entry.readable().getReader();
1110+
await reader.read();
1111+
await new Promise((resolve) => setTimeout(resolve, 50));
1112+
const readAhead = reads.reduce((a, b) => a + b, 0);
1113+
while (!(await reader.read()).done);
1114+
1115+
// A few queued slices are expected; the whole 1 MiB entry is not
1116+
expect(readAhead).toBeLessThan(compressed.length / 2);
1117+
expect(Math.max(...reads)).toBeLessThanOrEqual(65536);
1118+
});
11011119
});
11021120

11031121
describe("source closure lifecycle", () => {

0 commit comments

Comments
 (0)